diff --git a/.changeset/circular-object-hashing.md b/.changeset/circular-object-hashing.md new file mode 100644 index 00000000..cef64d69 --- /dev/null +++ b/.changeset/circular-object-hashing.md @@ -0,0 +1,5 @@ +--- +'@cleverbrush/deep': patch +--- + +Prevent circular object references from overflowing the stack during internal object hashing while preserving acyclic hash results. diff --git a/.changeset/close-rejected-subscriptions.md b/.changeset/close-rejected-subscriptions.md new file mode 100644 index 00000000..5d639d76 --- /dev/null +++ b/.changeset/close-rejected-subscriptions.md @@ -0,0 +1,6 @@ +--- +'@cleverbrush/server': patch +--- + +Close WebSocket subscriptions when middleware rejects the request without +calling the next handler, preventing idle unauthorized connections. diff --git a/.changeset/preserve-column-default-changes.md b/.changeset/preserve-column-default-changes.md new file mode 100644 index 00000000..af45c197 --- /dev/null +++ b/.changeset/preserve-column-default-changes.md @@ -0,0 +1,9 @@ +--- +'@cleverbrush/knex-schema': patch +--- + +Apply and generate column default changes without dropping, retyping or changing +the nullability of the existing column. Removing a default executes DROP DEFAULT; +setting one executes SET DEFAULT with safely quoted values (including question +marks and backslashes) and support for explicit SQL expressions. Generated down +migrations restore the original default instead of assuming a timestamp column. diff --git a/.changeset/subscription-empty-history.md b/.changeset/subscription-empty-history.md new file mode 100644 index 00000000..47e68729 --- /dev/null +++ b/.changeset/subscription-empty-history.md @@ -0,0 +1,5 @@ +--- +'@cleverbrush/client': patch +--- + +Respect maxEvents: 0 in useSubscription by retaining only lastEvent and leaving the event history empty. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index d8cbc897..61c71456 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -46,8 +46,11 @@ npm run build:docs-site floors for **unit** coverage. New published packages need explicit floors; do not lower existing floors to hide regressions. Refresh README badges explicitly with `npm run coverage:badges` after a successful coverage run. Badges do not include -the dedicated database or S3 integration suites. In particular, -`scheduler-postgres` relies on the real-database suite, not its small unit suite. +the dedicated database or S3 integration suites. Query, ORM and PostgreSQL +scheduler unit tests use the real Knex compiler with a simulated driver boundary +to check SQL, bindings, row decoding and failure paths. These do not prove database +locking, concurrent claims, lease recovery or transactional behavior: the real +PostgreSQL suites remain mandatory even when unit coverage reaches 100%. The package smoke test packs every published workspace, installs the tarballs and peer dependencies in a disposable consumer, checks every export and TypeScript diff --git a/README.md b/README.md index 50722066..9dd534f9 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ [![License: BSD-3-Clause](https://img.shields.io/badge/license-BSD--3--Clause-blue.svg)](./LICENSE) [![Standard Schema v1](https://img.shields.io/badge/Standard%20Schema-v1-blue)](https://standardschema.dev/) -![Unit coverage](https://img.shields.io/badge/unit_coverage-82.2%25-green) +![Unit coverage](https://img.shields.io/badge/unit_coverage-92.2%25-brightgreen) Cleverbrush is a schema-first TypeScript framework monorepo. It provides the diff --git a/coverage-thresholds.json b/coverage-thresholds.json index a4777d0f..81eef547 100644 --- a/coverage-thresholds.json +++ b/coverage-thresholds.json @@ -12,16 +12,16 @@ "lines": 92 }, "client": { - "statements": 83, - "branches": 79, - "functions": 89, - "lines": 84 + "statements": 93, + "branches": 88, + "functions": 96, + "lines": 95 }, "deep": { - "statements": 78, - "branches": 88, - "functions": 89, - "lines": 77 + "statements": 98, + "branches": 95, + "functions": 100, + "lines": 98 }, "di": { "statements": 94, @@ -42,10 +42,10 @@ "lines": 100 }, "knex-schema": { - "statements": 63, - "branches": 59, - "functions": 66, - "lines": 64 + "statements": 80, + "branches": 75, + "functions": 80, + "lines": 83 }, "log": { "statements": 98, @@ -54,22 +54,22 @@ "lines": 98 }, "mapper": { - "statements": 93, - "branches": 83, - "functions": 93, - "lines": 97 + "statements": 97, + "branches": 90, + "functions": 100, + "lines": 98 }, "orm": { - "statements": 80, - "branches": 61, - "functions": 86, - "lines": 83 + "statements": 90, + "branches": 80, + "functions": 97, + "lines": 94 }, "orm-cli": { - "statements": 59, - "branches": 60, - "functions": 60, - "lines": 59 + "statements": 90, + "branches": 85, + "functions": 100, + "lines": 98 }, "otel": { "statements": 88, @@ -78,22 +78,22 @@ "lines": 90 }, "react-form": { - "statements": 95, - "branches": 89, - "functions": 94, - "lines": 96 + "statements": 99, + "branches": 95, + "functions": 100, + "lines": 99 }, "scheduler": { - "statements": 83, - "branches": 82, - "functions": 89, - "lines": 85 + "statements": 95, + "branches": 90, + "functions": 97, + "lines": 97 }, "scheduler-postgres": { - "statements": 11, - "branches": 21, - "functions": 2, - "lines": 11 + "statements": 90, + "branches": 85, + "functions": 100, + "lines": 100 }, "schema": { "statements": 97, @@ -108,10 +108,10 @@ "lines": 97 }, "server": { - "statements": 81, - "branches": 78, - "functions": 85, - "lines": 82 + "statements": 90, + "branches": 85, + "functions": 94, + "lines": 92 }, "server-openapi": { "statements": 96, diff --git a/libs/client/README.md b/libs/client/README.md index 6484087a..ddbcef6b 100644 --- a/libs/client/README.md +++ b/libs/client/README.md @@ -1,6 +1,6 @@ # @cleverbrush/client -![Unit coverage](https://img.shields.io/badge/unit_coverage-84%25-green) +![Unit coverage](https://img.shields.io/badge/unit_coverage-94.6%25-brightgreen) Typed HTTP client for `@cleverbrush/server` API contracts — zero codegen, full type safety. Optional React + TanStack Query integration via `@cleverbrush/client/react`. @@ -699,7 +699,7 @@ function LiveFeed() { | Option | Type | Default | Description | |--------|------|---------|-------------| | `enabled` | `boolean` | `true` | Toggle the subscription on/off | -| `maxEvents` | `number` | unlimited | Maximum events to keep in the `events` array | +| `maxEvents` | `number` | unlimited | Maximum events to keep in the `events` array; `0` retains no history | ## React Integration (`@cleverbrush/client/react`) diff --git a/libs/client/src/client.transports.test.ts b/libs/client/src/client.transports.test.ts new file mode 100644 index 00000000..b3bb5540 --- /dev/null +++ b/libs/client/src/client.transports.test.ts @@ -0,0 +1,111 @@ +import { describe, expect, it, vi } from 'vitest'; +import { createClient } from './client.js'; +import { ApiError, NetworkError } from './errors.js'; + +function setup(meta: Record = {}, options = {}) { + const fetch = vi.fn(); + const ep = { + introspect: () => ({ + method: 'POST', + basePath: '/items', + pathTemplate: '', + ...meta + }) + }; + const client = createClient({ items: { action: ep } } as any, { + fetch, + ...options + }) as any; + return { fetch, call: client.items.action }; +} +describe('HTTP response and upload transports', () => { + it('encodes mixed multipart files and removes caller content-type boundaries', async () => { + const { fetch, call } = setup( + { fileUpload: {} }, + { headers: { 'Content-Type': 'application/json' } } + ); + fetch.mockImplementation( + async () => + new Response('{}', { + headers: { 'content-type': 'application/json' } + }) + ); + const blob = new Blob(['one'], { type: 'text/plain' }); + await call({ + body: { title: 'report', absent: undefined }, + files: { + attachment: [ + blob, + { + buffer: new Uint8Array([2]), + filename: 'two.txt', + mimeType: 'text/plain' + }, + undefined + ] + }, + headers: { 'content-type': 'bad' }, + retry: false, + timeout: 100, + optimisticUpdate: {}, + offlineQueue: true + }); + const init = fetch.mock.calls[0][1]; + expect(new Headers(init.headers).has('content-type')).toBe(false); + expect(init.body.get('title')).toBe('report'); + expect(init.body.has('absent')).toBe(false); + expect(init.body.getAll('attachment')).toHaveLength(2); + expect(init.body.getAll('attachment')[1].name).toBe('two.txt'); + await call(); + expect(fetch.mock.calls[1][1].body).toBeInstanceOf(FormData); + }); + it('normalizes non-Error failures and runs error hooks for existing web errors', async () => { + const hook = vi.fn(e => e); + const { fetch, call } = setup({}, { hooks: { beforeError: [hook] } }); + fetch.mockRejectedValueOnce('network string'); + await expect(call()).rejects.toThrow('network string'); + const error = new NetworkError('already wrapped'); + fetch.mockRejectedValueOnce(error); + await expect(call()).rejects.toBe(error); + expect(hook).toHaveBeenCalledWith(error); + }); + it('returns text successes and supports file downloads and unauthorized failures', async () => { + const unauthorized = vi.fn(); + const { fetch, call } = setup({}, { onUnauthorized: unauthorized }); + fetch.mockResolvedValueOnce(new Response('plain text')); + expect(await call()).toBe('plain text'); + fetch.mockResolvedValueOnce(new Response('file bytes')); + expect(await (await call.file()).text()).toBe('file bytes'); + fetch.mockResolvedValueOnce(new Response(null, { status: 401 })); + await expect(call.file()).rejects.toBeInstanceOf(ApiError); + expect(unauthorized).toHaveBeenCalledOnce(); + }); + it('streams incomplete chunks, supports empty bodies and translates failures', async () => { + const unauthorized = vi.fn(); + const beforeError = vi.fn(e => e); + const { fetch, call } = setup( + {}, + { + onUnauthorized: unauthorized, + hooks: { beforeError: [beforeError] } + } + ); + fetch.mockResolvedValueOnce(new Response('one\ntwo\nlast')); + const lines = []; + for await (const line of call.stream()) lines.push(line); + expect(lines).toEqual(['one', 'two', 'last']); + fetch.mockResolvedValueOnce(new Response(null, { status: 204 })); + const empty = []; + for await (const line of call.stream()) empty.push(line); + expect(empty).toEqual([]); + fetch.mockResolvedValueOnce(new Response(null, { status: 401 })); + await expect( + call.stream()[Symbol.asyncIterator]().next() + ).rejects.toBeInstanceOf(ApiError); + expect(unauthorized).toHaveBeenCalledOnce(); + fetch.mockRejectedValueOnce(new TypeError('offline')); + await expect( + call.stream()[Symbol.asyncIterator]().next() + ).rejects.toBeInstanceOf(NetworkError); + }); +}); diff --git a/libs/client/src/react/hooks.factories.test.ts b/libs/client/src/react/hooks.factories.test.ts new file mode 100644 index 00000000..5d5458e2 --- /dev/null +++ b/libs/client/src/react/hooks.factories.test.ts @@ -0,0 +1,114 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { + createPrefetch, + createUseInfiniteQuery, + createUseMutation, + createUseQuery, + createUseSuspenseQuery +} from './hooks.js'; + +const mocks = vi.hoisted(() => ({ + query: vi.fn(o => o), + suspense: vi.fn(o => o), + infinite: vi.fn(o => o), + mutation: vi.fn(o => o), + invalidate: vi.fn() +})); +vi.mock('@tanstack/react-query', () => ({ + useQuery: mocks.query, + useSuspenseQuery: mocks.suspense, + useInfiniteQuery: mocks.infinite, + useMutation: mocks.mutation, + useQueryClient: () => ({ invalidateQueries: mocks.invalidate }) +})); +describe('endpoint hook factories', () => { + const call = vi.fn().mockResolvedValue('response'); + const client = { items: { list: call, create: call } }; + beforeEach(() => vi.clearAllMocks()); + it.each([ + undefined, + null, + 3, + { query: { page: 1 } }, + ...[ + 'enabled', + 'staleTime', + 'gcTime', + 'refetchInterval', + 'retry', + 'select', + 'placeholderData' + ].map(key => ({ [key]: false })) + ])('distinguishes query arguments from options: %j', async value => { + const hook = createUseQuery(client, 'items', 'list'); + const options = hook(value); + expect(await options.queryFn()).toBe('response'); + const isOptions = + value && typeof value === 'object' && !('query' in value); + expect(call).toHaveBeenCalledWith(isOptions ? undefined : value); + expect(options.queryKey.slice(0, 3)).toEqual([ + '@cleverbrush', + 'items', + 'list' + ]); + }); + it('executes suspense queries, argument queries and paginated query functions', async () => { + const args = { query: { page: 2 } }; + await createUseQuery( + client, + 'items', + 'list' + )(args, { enabled: true }).queryFn(); + expect(call).toHaveBeenLastCalledWith(args); + await createUseSuspenseQuery( + client, + 'items', + 'list' + )(args, { staleTime: 100 }).queryFn(); + expect(call).toHaveBeenLastCalledWith(args); + await createUseSuspenseQuery( + client, + 'items', + 'list' + )({ staleTime: 100 }).queryFn(); + expect(call).toHaveBeenLastCalledWith(undefined); + await createUseInfiniteQuery( + client, + 'items', + 'list' + )(page => ({ query: { page } }), { initialPageParam: 1 }).queryFn({ + pageParam: 3 + }); + expect(call).toHaveBeenLastCalledWith({ query: { page: 3 } }); + }); + it('invalidates only tagged mutations and preserves success callbacks', async () => { + const success = vi.fn(); + const hook = createUseMutation(client, 'items', 'create', ['items']); + const options = hook({ onSuccess: success }); + await options.mutationFn({ body: { name: 'new' } }); + options.onSuccess('saved', {}, undefined); + expect(success).toHaveBeenCalledWith('saved', {}, undefined); + expect(mocks.invalidate).toHaveBeenCalledWith({ + queryKey: ['@cleverbrush', 'items'] + }); + mocks.invalidate.mockClear(); + createUseMutation(client, 'items', 'create')().onSuccess(); + createUseMutation(client, 'items', 'create', [])().onSuccess(); + expect(mocks.invalidate).not.toHaveBeenCalled(); + }); + it('prefetches using the same query arguments and key', async () => { + const prefetchQuery = vi.fn(async opts => { + await opts.queryFn(); + }); + await createPrefetch( + client, + 'items', + 'list' + )({ prefetchQuery }, { query: { page: 2 } }); + expect(call).toHaveBeenCalledWith({ query: { page: 2 } }); + expect(prefetchQuery.mock.calls[0][0].queryKey).toEqual( + createUseQuery(client, 'items', 'list')({ query: { page: 2 } }) + .queryKey + ); + }); +}); diff --git a/libs/client/src/react/useSubscription.test.ts b/libs/client/src/react/useSubscription.test.ts new file mode 100644 index 00000000..2a13ba53 --- /dev/null +++ b/libs/client/src/react/useSubscription.test.ts @@ -0,0 +1,124 @@ +import { act, cleanup, renderHook } from '@testing-library/react'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import type { Subscription } from '../types.js'; +import { useSubscription } from './useSubscription.js'; + +class Feed implements Subscription { + state: Subscription['state'] = 'connecting'; + private resolve?: (result: IteratorResult) => void; + private reject?: (error: unknown) => void; + send = vi.fn(); + close = vi.fn(() => { + this.state = 'closed'; + this.resolve?.({ done: true, value: undefined }); + }); + emit(value: number) { + this.resolve?.({ done: false, value }); + } + fail(error: unknown) { + this.reject?.(error); + } + [Symbol.asyncIterator](): AsyncIterator { + return { + next: () => + new Promise((resolve, reject) => { + this.resolve = resolve; + this.reject = reject; + }) + }; + } +} +afterEach(() => { + cleanup(); + vi.useRealTimers(); +}); + +describe('subscription hook lifecycle', () => { + it('tracks connection state, retains bounded events and forwards send/close', async () => { + vi.useFakeTimers(); + const feed = new Feed(); + const subscribe = vi.fn(() => feed); + const { result, rerender, unmount } = renderHook( + ({ maxEvents }) => useSubscription(subscribe, { maxEvents }), + { initialProps: { maxEvents: 2 } } + ); + expect(result.current.state).toBe('connecting'); + act(() => { + feed.state = 'connected'; + vi.advanceTimersByTime(400); + }); + expect(result.current.state).toBe('connected'); + for (const value of [1, 2, 3]) await act(async () => feed.emit(value)); + expect(result.current.events).toEqual([2, 3]); + expect(result.current.lastEvent).toBe(3); + rerender({ maxEvents: 1 }); + await act(async () => feed.emit(4)); + expect(result.current.events).toEqual([4]); + expect(subscribe).toHaveBeenCalledTimes(1); + act(() => result.current.send('message')); + expect(feed.send).toHaveBeenCalledExactlyOnceWith('message'); + await act(async () => result.current.close()); + expect(result.current.state).toBe('closed'); + act(() => result.current.send('ignored')); + expect(feed.send).toHaveBeenCalledTimes(1); + unmount(); + expect(vi.getTimerCount()).toBe(0); + }); + + it('supports disabled hooks, re-enabling and cleanup', async () => { + vi.useFakeTimers(); + const feeds: Feed[] = []; + const subscribe = vi.fn(() => { + const feed = new Feed(); + feeds.push(feed); + return feed; + }); + const { result, rerender, unmount } = renderHook( + ({ enabled }) => useSubscription(subscribe, { enabled }), + { initialProps: { enabled: false } } + ); + act(() => { + result.current.send('ignored'); + result.current.close(); + }); + expect(subscribe).not.toHaveBeenCalled(); + rerender({ enabled: true }); + await act(async () => feeds[0].emit(1)); + rerender({ enabled: false }); + expect(feeds[0].close).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + rerender({ enabled: true }); + expect(subscribe).toHaveBeenCalledTimes(2); + unmount(); + expect(feeds[1].close).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + }); + + it.each([new Error('disconnect'), 'disconnect'])( + 'normalizes iterator failure %s and closes state', + async error => { + const feed = new Feed(); + const { result } = renderHook(() => useSubscription(() => feed)); + await act(async () => feed.fail(error)); + expect(result.current.error).toEqual(new Error('disconnect')); + expect(result.current.state).toBe('closed'); + } + ); + + it('keeps the latest event without retaining history when maxEvents is zero', async () => { + const feed = new Feed(); + const { result } = renderHook(() => + useSubscription(() => feed, { maxEvents: 0 }) + ); + await act(async () => feed.emit(5)); + expect(result.current.lastEvent).toBe(5); + expect(result.current.events).toEqual([]); + }); + + it('retains unlimited history when no maximum is configured', async () => { + const feed = new Feed(); + const { result } = renderHook(() => useSubscription(() => feed)); + for (const value of [1, 2, 3]) await act(async () => feed.emit(value)); + expect(result.current.events).toEqual([1, 2, 3]); + }); +}); diff --git a/libs/client/src/react/useSubscription.ts b/libs/client/src/react/useSubscription.ts index 0cdaef14..e08f2e8c 100644 --- a/libs/client/src/react/useSubscription.ts +++ b/libs/client/src/react/useSubscription.ts @@ -119,7 +119,7 @@ export function useSubscription( const next = [...prev, event]; const max = maxEventsRef.current; if (max !== undefined && next.length > max) { - return next.slice(-max); + return max === 0 ? [] : next.slice(-max); } return next; }); diff --git a/libs/deep/README.md b/libs/deep/README.md index e473daa0..f8c20a5e 100644 --- a/libs/deep/README.md +++ b/libs/deep/README.md @@ -3,7 +3,7 @@ [![CI](https://github.com/cleverbrush/framework/actions/workflows/ci.yml/badge.svg)](https://github.com/cleverbrush/framework/actions/workflows/ci.yml) [![License: BSD-3-Clause](https://img.shields.io/badge/license-BSD--3--Clause-blue.svg)](../../LICENSE) -![Unit coverage](https://img.shields.io/badge/unit_coverage-78.9%25-yellowgreen) +![Unit coverage](https://img.shields.io/badge/unit_coverage-98.7%25-brightgreen) A library for deep operations on JavaScript objects — cloning, equality, merging, and flattening. diff --git a/libs/deep/src/deepExtend.test.ts b/libs/deep/src/deepExtend.test.ts index 490d45b4..a0112ed7 100644 --- a/libs/deep/src/deepExtend.test.ts +++ b/libs/deep/src/deepExtend.test.ts @@ -258,3 +258,8 @@ test('deepExtend does not recurse into inherited target properties', () => { function removePollutedMarker(): void { delete (Object.prototype as { polluted?: unknown }).polluted; } +test('rejects an empty merge and invalid first sources', () => { + expect(() => deepExtend()).toThrow('no arguments'); + expect(() => deepExtend(null)).toThrow('not a non-null object'); + expect(() => deepExtend('value')).toThrow('not a non-null object'); +}); diff --git a/libs/deep/src/hashObject.test.ts b/libs/deep/src/hashObject.test.ts new file mode 100644 index 00000000..46e1b8e7 --- /dev/null +++ b/libs/deep/src/hashObject.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from 'vitest'; +import { HashObject } from './hashObject.js'; + +describe('object hashing', () => { + it('keeps numeric inputs and hashes strings deterministically', () => { + expect(HashObject(42)).toBe(42); + expect(HashObject('hello')).toBe(-1502471327); + expect(HashObject('42')).not.toBe(HashObject(42)); + expect(HashObject('')).toBe(-891985903); + expect(HashObject(undefined)).toBeTypeOf('number'); + expect(HashObject(NaN)).toBeTypeOf('number'); + }); + + it('sorts keys and includes nested values without modifying them', () => { + const first = { a: 1, b: { label: 'one' } }; + const second = { b: { label: 'one' }, a: 1 }; + expect(HashObject(first)).toBe(HashObject(second)); + // Golden value from the pre-cycle-fix implementation. + expect(HashObject(first)).toBe(1868467459); + expect(HashObject({ one: 'value', two: 'value' })).toBe(538503302); + expect(HashObject(first)).not.toBe(HashObject({ ...first, a: 2 })); + expect(first).toEqual({ a: 1, b: { label: 'one' } }); + expect(HashObject({})).toBe(0); + }); + + it('terminates self and mutual references with stable hashes', () => { + const self: any = { label: 'one' }; + self.self = self; + const other: any = { label: 'one' }; + other.self = other; + expect(HashObject(self)).toBeTypeOf('number'); + expect(HashObject(self)).toBe(HashObject(other)); + const left: any = { label: 'left' }; + const right: any = { label: 'right', left }; + left.right = right; + expect(HashObject(left)).toBe(HashObject(left)); + expect(left.right.left).toBe(left); + }); + + it('respects explicit exclusions and shared references', () => { + const shared = { value: 'same' }; + expect(HashObject({ a: shared, b: shared })).toBe( + HashObject({ b: shared, a: shared }) + ); + expect(HashObject(shared, [shared])).toBeUndefined(); + }); +}); diff --git a/libs/deep/src/hashObject.ts b/libs/deep/src/hashObject.ts index af3dda75..36506d89 100644 --- a/libs/deep/src/hashObject.ts +++ b/libs/deep/src/hashObject.ts @@ -14,48 +14,56 @@ const stringHash = (string: string, noType?: boolean) => { function objectHash( obj: Record, - exclude: any[] + exclude: any[], + ancestors: WeakSet ): number | undefined { - if (exclude.indexOf(obj) > -1) { + if (exclude.indexOf(obj) > -1 || ancestors.has(obj)) { return undefined; } - let hash = ''; - const keys = Object.keys(obj).sort(); - for (let index = 0; index < keys.length; index += 1) { - const key = keys[index]; - const keyHash = hashFunc(key); - const attrHash = hashFunc(obj[key], exclude); - exclude.push(obj[key]); - hash += stringHash(`object${keyHash}${attrHash}`, true); + ancestors.add(obj); + try { + let hash = ''; + const keys = Object.keys(obj).sort(); + for (const key of keys) { + const keyHash = hashValue(key, [], ancestors); + const attrHash = hashValue(obj[key], exclude, ancestors); + exclude.push(obj[key]); + hash += stringHash(`object${keyHash}${attrHash}`, true); + } + return stringHash(hash, true); + } finally { + ancestors.delete(obj); } - return stringHash(hash, true); } /** * Computes a 32-bit numeric hash for any value. * * Objects are hashed by recursively hashing their sorted keys and values; - * circular references are detected via an internal exclusion list. + * Circular object references terminate at the repeated ancestor. Existing + * acyclic hashes and the supplied exclusion list retain their behavior. * * @param unkType - The value to hash (object, string, number, etc.). * @param exclude - Internal array used for circular-reference detection. * @returns A 32-bit integer hash code. */ export function HashObject(unkType: any, exclude?: any[]): number { - let ex = exclude; - if (ex === undefined) { - ex = []; - } + return hashValue(unkType, exclude ?? [], new WeakSet()); +} + +function hashValue( + unkType: any, + exclude: any[], + ancestors: WeakSet +): number { // biome-ignore lint/suspicious/noGlobalIsNan: intentional coercion — isNaN returns true for non-numeric types like objects if (!isNaN(unkType) && typeof unkType !== 'string') { return unkType; } switch (typeof unkType) { case 'object': - return objectHash(unkType, ex) as number; + return objectHash(unkType, exclude, ancestors) as number; default: return stringHash(String(unkType)) as number; } } - -const hashFunc = HashObject; diff --git a/libs/knex-schema/README.md b/libs/knex-schema/README.md index d6b66977..857e6547 100644 --- a/libs/knex-schema/README.md +++ b/libs/knex-schema/README.md @@ -1,6 +1,6 @@ # @cleverbrush/knex-schema -![Unit coverage](https://img.shields.io/badge/unit_coverage-63.5%25-yellow) +![Unit coverage](https://img.shields.io/badge/unit_coverage-83%25-green) Type-safe, schema-driven query builder for [Knex](https://knexjs.org/). Use `@cleverbrush/schema` object builders to describe your PostgreSQL tables — column name mapping, eager loading, and full CRUD are handled automatically with complete TypeScript inference. @@ -636,8 +636,8 @@ See the `@cleverbrush/orm` docs for the full inheritance API (`.ofVariant()` etc | `entitiesToSnapshot(entities)` | Materialise entity definitions into a JSON-serialisable schema snapshot | | `loadSnapshot(path)` / `writeSnapshot(path, snap)` | Read/write the committed snapshot file | | `generateMigrationsForContext(entities, prevSnapshot)` | Diff entities against the snapshot and emit a TS migration source plus the next snapshot | -| `generateMigration(snapshotA, snapshotB)` | Lower-level snapshot-vs-snapshot diff | -| `diffSchema(schema, dbState)` / `applyDiff(knex, diff, table)` | Live-database diff/apply (used by `cb-orm db push`) | +| `generateMigration(diff, tableName)` | Emit up/down migration source from a `MigrationDiff`; default changes preserve the column's type and nullability | +| `diffSchema(schema, dbState)` / `applyDiff(knex, diff, table)` | Live-database diff/apply (used by `cb-orm db push`); default changes preserve column types and stored values | | `validateEntitiesAgainstDatabase(knex, entities)` | Read-only live-database validation for CI drift checks | | `introspectDatabase(knex, table)` / `tableExistsInDb(knex, table)` | Database introspection helpers | | `generateCreateTable(schema)` / `generateCreatePolymorphicTables(schema)` | Knex-statement builders for fresh `CREATE TABLE` | diff --git a/libs/knex-schema/integration/migrations.test.ts b/libs/knex-schema/integration/migrations.test.ts new file mode 100644 index 00000000..32b47ee5 --- /dev/null +++ b/libs/knex-schema/integration/migrations.test.ts @@ -0,0 +1,80 @@ +import { randomUUID } from 'node:crypto'; +import { + applyDiff, + generateMigration, + type MigrationDiff +} from '@cleverbrush/knex-schema'; +import Knex from 'knex'; +import { afterAll, afterEach, beforeEach, expect, it } from 'vitest'; + +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 }); +const table = `cb_defaults_${randomUUID().replaceAll('-', '')}`; +beforeEach(async () => { + await knex.schema.createTable(table, t => { + t.increments('id'); + t.text('status').notNullable().defaultTo('draft'); + }); + await knex(table).insert({ status: 'preserved' }); +}); +afterEach(() => knex.schema.dropTableIfExists(table)); +afterAll(() => knex.destroy()); +function change(value: unknown): MigrationDiff { + return { + addColumns: [], + dropColumns: [], + addIndexes: [], + dropIndexes: [], + addForeignKeys: [], + dropForeignKeys: [], + alterColumns: [ + { + name: 'status', + changes: { defaultValue: { from: 'draft', to: value } } + } + ] + }; +} +it('changes and removes defaults without retyping columns or losing stored data', async () => { + await applyDiff(knex, change("reviewer's draft?"), table); + await knex(table).insert({}); + expect((await knex(table).orderBy('id')).map(row => row.status)).toEqual([ + 'preserved', + "reviewer's draft?" + ]); + await applyDiff(knex, change({ raw: "upper('pending')" }), table); + await knex(table).insert({}); + expect((await knex(table).orderBy('id', 'desc').first()).status).toBe( + 'PENDING' + ); + await applyDiff(knex, change(null), table); + const columns = await knex(table).columnInfo(); + expect(columns.status.type).toBe('text'); + expect(columns.status.defaultValue).toBeNull(); + expect((await knex(table).where('id', 1).first()).status).toBe('preserved'); + await expect(knex(table).insert({})).rejects.toThrow(/not-null/); +}); + +it('executes generated default up/down migrations and preserves literal values', async () => { + const value = "reviewer's \\? draft\n'; DROP TABLE ignored; --"; + const migration = generateMigration(change(value), table); + const execute = (body: string) => + new Function('knex', `return (async () => { ${body} })();`)(knex); + await execute(migration.up); + await knex(table).insert({}); + expect((await knex(table).orderBy('id', 'desc').first()).status).toBe( + value + ); + await execute(migration.down); + await knex(table).insert({}); + expect((await knex(table).orderBy('id', 'desc').first()).status).toBe( + 'draft' + ); + await applyDiff(knex, change(value), table); + await knex(table).insert({}); + expect((await knex(table).orderBy('id', 'desc').first()).status).toBe( + value + ); + expect((await knex(table).columnInfo()).status.type).toBe('text'); +}); diff --git a/libs/knex-schema/src/aliased-query.test.ts b/libs/knex-schema/src/aliased-query.test.ts new file mode 100644 index 00000000..13dbf787 --- /dev/null +++ b/libs/knex-schema/src/aliased-query.test.ts @@ -0,0 +1,72 @@ +import Knex from 'knex'; +import { afterAll, describe, expect, it } from 'vitest'; +import { AliasedQuerySource, alias, and, eq, or } from './aliased-query.js'; +import { aggregate } from './expressions.js'; +import { number, object, string } from './extension.js'; + +const Schema = object({ + id: number().primaryKey(), + name: string().hasColumnName('display_name') +}).hasTableName('items'); +const knex = Knex({ client: 'pg' }); +afterAll(() => knex.destroy()); +const source = () => new AliasedQuerySource(knex, alias(Schema, 'a')); +describe('flat join SQL planner', () => { + it('compiles nested join predicates, mapped filtering, grouping, ordering and aggregate selection', () => { + const q = source() + .join(alias(Schema, 'b'), t => + and( + eq(t.a.id, t.b.id), + or(eq(t.a.name, t.b.name), eq(t.a.id, t.b.id)) + ) + ) + .leftJoin(alias(Schema, 'c'), t => eq(t.a.id, t.c.id)) + .where(t => t.a.name, 'like', '%a%') + .whereIn(t => t.a.id, [1, 2]) + .whereNull(t => t.c.name) + .whereNotNull(t => t.a.name) + .groupBy(t => t.a.name) + .having(t => t.a.id, '>', 0) + .select(t => ({ name: t.a.name, count: aggregate.count(t.a.id) })) + .orderBy(t => t.a.name, 'desc') + .orderByRaw('?? asc', ['a.id']) + .limit(2) + .offset(1); + expect(q.toQuery()).toContain('left join'); + expect(q.toQuery()).toContain('having'); + expect(q.toQuery()).toContain('count('); + expect(q.toQuery()).toContain('"a"."display_name"'); + expect(q.cloneReadSource().toQuery()).toBe(q.toQuery()); + }); + it('rejects empty/foreign predicates, duplicate aliases and unsupported operators', () => { + expect(() => and()).toThrow('at least one'); + expect(() => or()).toThrow('at least one'); + expect(() => + source().join(alias(Schema, 'a'), t => eq(t.a.id, t.a.id)) + ).toThrow(); + expect(() => + source().join(alias(Schema, 'b'), () => ({}) as any) + ).toThrow(); + expect(() => source().where(() => ({}) as any, 1)).toThrow('column'); + expect(() => source().where(t => t.a.id, 'unsafe', 1)).toThrow(); + expect(() => source().having(t => t.a.id, 'unsafe', 1)).toThrow(); + expect( + source() + .orderBy(t => t.a.id, 'unsafe' as any) + .select(t => ({ id: t.a.id })) + .toQuery() + ).toContain('asc'); + expect(() => source().select(() => ({ nope: 1 }) as any)).toThrow(); + const plain = source() + .where(t => t.a.id, 1) + .select(t => ({ id: t.a.id })); + expect(plain.toQuery()).toContain('= 1'); + expect( + plain + .apply(q => { + q.whereRaw('true'); + }) + .toQuery() + ).toContain('true'); + }); +}); diff --git a/libs/knex-schema/src/migration.execution.test.ts b/libs/knex-schema/src/migration.execution.test.ts new file mode 100644 index 00000000..a786e091 --- /dev/null +++ b/libs/knex-schema/src/migration.execution.test.ts @@ -0,0 +1,255 @@ +import { afterEach, describe, expect, it } from 'vitest'; +import { mockDriver } from '../testing/mock-driver.js'; +import { + applyDiff, + generateMigration, + introspectDatabase +} from './migration.js'; +import type { MigrationDiff } from './types.js'; + +const connections: ReturnType[] = []; +function setup() { + const d = mockDriver(); + connections.push(d); + return d; +} +const empty = (): MigrationDiff => ({ + addColumns: [], + dropColumns: [], + alterColumns: [], + addIndexes: [], + dropIndexes: [], + addForeignKeys: [], + dropForeignKeys: [] +}); +afterEach(async () => { + await Promise.all(connections.splice(0).map(d => d.knex.destroy())); +}); +describe('migration execution', () => { + it('normalizes PostgreSQL catalog rows without interpolating table names', async () => { + const d = setup(); + d.respond = q => + q.sql.includes('information_schema.columns') + ? [ + { + column_name: 'name', + data_type: 'text', + is_nullable: 'YES' + } + ] + : q.sql.includes('pg_indexes') + ? [ + { + indexname: 'unique_name', + indexdef: 'CREATE UNIQUE INDEX', + columns: ['name'] + }, + { indexname: 'plain' } + ] + : q.sql.includes('referential_constraints') + ? [ + { + constraint_name: 'owner_fk', + column_name: 'owner_id', + foreign_table: 'owners', + foreign_column: 'id', + delete_rule: 'CASCADE', + update_rule: 'RESTRICT' + } + ] + : [ + { + conname: 'length_check', + definition: 'CHECK (length(name) > 0)' + } + ]; + const state = await introspectDatabase( + d.knex, + "items'; DROP TABLE items; --" + ); + expect(state.columns.name).toMatchObject({ + nullable: true, + defaultValue: null, + maxLength: null + }); + expect(state.indexes).toMatchObject([ + { columns: ['name'], unique: true }, + { columns: [], unique: false } + ]); + expect(state.foreignKeys[0].foreignTable).toBe('owners'); + expect(state.checks[0].name).toBe('length_check'); + for (const q of d.queries) { + expect(q.sql).not.toContain('DROP TABLE'); + expect(q.bindings).toContain("items'; DROP TABLE items; --"); + } + }); + it('applies column, index and foreign-key changes in the intended order', async () => { + const d = setup(); + const diff = empty(); + diff.addColumns = [ + 'integer', + 'character varying', + 'text', + 'boolean', + 'timestamp', + 'timestamp without time zone', + 'double precision', + 'float', + 'jsonb', + 'json', + 'uuid', + 'numeric(12,2)' + ].map((type, i) => ({ + name: `c${i}`, + type, + nullable: i % 2 === 0, + defaultValue: + i === 0 + ? 'now' + : i === 1 + ? { raw: "'draft'" } + : i === 2 + ? 'value' + : undefined + })); + diff.addColumns.push({ + name: 'owner_id', + type: 'integer', + nullable: false, + references: { table: 'owners', column: 'id' }, + onDelete: 'CASCADE', + onUpdate: 'CASCADE' + }); + diff.dropColumns = ['obsolete']; + diff.alterColumns = [ + { name: 'name', changes: { nullable: { from: false, to: true } } }, + { name: 'status', changes: { nullable: { from: true, to: false } } } + ]; + diff.addIndexes = [ + { columns: ['c0'], unique: true, name: 'unique_c0' }, + { columns: ['c1'], unique: true }, + { columns: ['c2'], unique: false, name: 'index_c2' } + ]; + diff.dropIndexes = ['old_index']; + diff.dropForeignKeys = ['old_fk']; + diff.addForeignKeys = [ + { + column: 'c0', + foreignTable: 'owners', + foreignColumn: 'id', + onDelete: 'CASCADE', + onUpdate: 'RESTRICT' + } + ]; + await applyDiff(d.knex, diff, 'items'); + expect(d.queries[0].sql).toContain('DROP CONSTRAINT "old_fk"'); + const sql = d.queries.map(q => q.sql).join('\n'); + expect(sql).toContain('drop column "obsolete"'); + expect(sql).toContain('alter column "name" drop not null'); + expect(sql).toContain('alter column "status" set not null'); + expect(sql).toContain('on delete CASCADE'); + expect(sql).toContain('add constraint "unique_c0" unique'); + expect(sql).toContain('drop index "old_index"'); + expect(sql).toContain('numeric(12,2)'); + const before = d.queries.length; + await applyDiff(d.knex, empty(), 'items'); + expect(d.queries).toHaveLength(before); + await applyDiff( + d.knex, + { ...empty(), dropForeignKeys: ['another_fk'] }, + 'items' + ); + expect(d.queries).toHaveLength(before + 1); + }); + it('removes a default without deleting its column or existing data', async () => { + const d = setup(); + await applyDiff( + d.knex, + { + ...empty(), + alterColumns: [ + { + name: 'status', + changes: { defaultValue: { from: 'draft', to: null } } + } + ] + }, + 'items' + ); + expect(d.queries.map(q => q.sql).join('\n')).toContain( + 'ALTER COLUMN "status" DROP DEFAULT' + ); + expect(d.queries.map(q => q.sql).join('\n')).not.toContain( + 'drop column' + ); + }); + it('quotes default values while supporting explicit SQL defaults', async () => { + const d = setup(); + for (const value of [ + "reviewer's draft", + false, + 0, + 'now', + { raw: 'CURRENT_DATE' } + ]) { + await applyDiff( + d.knex, + { + ...empty(), + alterColumns: [ + { + name: 'status', + changes: { defaultValue: { from: null, to: value } } + } + ] + }, + 'items' + ); + } + expect(d.queries[0].sql).toContain("'reviewer''s draft'"); + expect(d.queries[0].bindings).toEqual([]); + expect(d.queries[1].sql).toContain('DEFAULT false'); + expect(d.queries[2].sql).toContain('DEFAULT 0'); + expect(d.queries[3].sql).toContain('CURRENT_TIMESTAMP'); + expect(d.queries[4].sql).toContain('CURRENT_DATE'); + }); + + it('generates reversible default DDL without altering column types or nullability', async () => { + const d = setup(); + for (const value of [ + null, + undefined, + "reviewer's draft?\\path", + 0, + false, + 'now', + { raw: 'CURRENT_DATE' } + ]) { + const diff: MigrationDiff = { + ...empty(), + alterColumns: [ + { + name: 'status', + changes: { defaultValue: { from: 'draft', to: value } } + } + ] + }; + const migration = generateMigration(diff, 'items'); + expect(migration.up).not.toContain('table.timestamp'); + expect(migration.down).not.toContain('.notNullable'); + const start = d.queries.length; + await new Function( + 'knex', + `return (async () => { ${migration.up} })();` + )(d.knex); + await new Function( + 'knex', + `return (async () => { ${migration.down} })();` + )(d.knex); + expect(d.queries[start].sql).toContain( + value == null ? 'DROP DEFAULT' : 'SET DEFAULT' + ); + expect(d.queries[start + 1].sql).toContain("SET DEFAULT U&'draft'"); + } + }); +}); diff --git a/libs/knex-schema/src/migration.ts b/libs/knex-schema/src/migration.ts index 0a7406cd..ccc67f0c 100644 --- a/libs/knex-schema/src/migration.ts +++ b/libs/knex-schema/src/migration.ts @@ -685,6 +685,7 @@ export function isDiffEmpty(diff: MigrationDiff): boolean { * * Foreign-key constraint drops are executed as raw `ALTER TABLE … DROP * CONSTRAINT` statements before the main `alterTable` call. + * Default changes preserve the column's type, nullability and stored data. * * @param knex - A configured Knex instance or transaction. * @param diff - The diff from {@link diffSchema}. @@ -713,6 +714,40 @@ export async function applyDiff( ]); } + // A default change must never recreate, retype or drop the data column. + for (const column of diff.alterColumns) { + const change = column.changes.defaultValue; + if (!change) continue; + if (change.to === null || change.to === undefined) { + await knex.raw('ALTER TABLE ?? ALTER COLUMN ?? DROP DEFAULT', [ + tableName, + column.name + ]); + } else { + const value = change.to; + const expression = + value === 'now' + ? knex.fn.now() + : typeof value === 'string' + ? knex.raw(defaultStringLiteral(value)) + : typeof value === 'object' && value !== null && value.raw + ? knex.raw(value.raw) + : value; + // PostgreSQL utility statements cannot use protocol parameters. + // Let Knex quote literals/identifiers, then execute the resulting DDL. + await knex.raw( + knex + .raw('ALTER TABLE ?? ALTER COLUMN ?? SET DEFAULT ?', [ + tableName, + column.name, + expression + ]) + .toQuery() + .replaceAll('?', '\\?') + ); + } + } + const hasTableChanges = diff.addColumns.length > 0 || diff.dropColumns.length > 0 || @@ -769,13 +804,6 @@ export async function applyDiff( } else { (table as any).dropNullable(col.name); } - } else if (key === 'defaultValue') { - if (change.to === null) { - table.dropColumn(col.name); // fallback — handled in source gen - } else { - // ALTER COLUMN SET DEFAULT is not directly in Knex builder; - // handled via raw in source generation - } } } } @@ -991,6 +1019,8 @@ function buildAlterTableFragments( ): { up: string; down: string } { const upLines: string[] = []; const downLines: string[] = []; + const upDefaults: string[] = []; + const downDefaults: string[] = []; // --- ADD COLUMNS --- for (const col of diff.addColumns) { @@ -1032,25 +1062,12 @@ function buildAlterTableFragments( downLines.push(` table.setNullable('${col.name}');`); } } else if (key === 'defaultValue') { - const toVal = change.to; - const fromVal = change.from; - if (toVal === null) { - upLines.push( - ` table.timestamp('${col.name}').alter(); // DROP DEFAULT` - ); - downLines.push( - ` table.timestamp('${col.name}').defaultTo(knex.fn.now()).alter();` - ); - } else { - upLines.push( - ` table.timestamp('${col.name}').notNullable().defaultTo(knex.fn.now()).alter();` - ); - downLines.push( - fromVal === null - ? ` table.timestamp('${col.name}').notNullable().alter();` - : ` table.timestamp('${col.name}').notNullable().defaultTo(knex.fn.now()).alter();` - ); - } + upDefaults.push( + defaultChangeCode(tableName, col.name, change.to) + ); + downDefaults.push( + defaultChangeCode(tableName, col.name, change.from) + ); } } } @@ -1098,17 +1115,48 @@ function buildAlterTableFragments( upLines.push(` // TODO: drop foreign key constraint '${fk}'`); } - const isEmpty = upLines.length === 0; + const up = + [ + ...upDefaults, + ...(upLines.length + ? [ + ` await knex.schema.alterTable('${tableName}', (table) => {\n${upLines.join('\n')}\n });` + ] + : []) + ].join('\n') || ' // No changes needed'; + const down = + [ + ...downDefaults, + ...(downLines.length + ? [ + ` await knex.schema.alterTable('${tableName}', (table) => {\n${downLines.join('\n')}\n });` + ] + : []) + ].join('\n') || ' // No changes needed'; - const up = isEmpty - ? ` // No changes needed` - : ` await knex.schema.alterTable('${tableName}', (table) => {\n${upLines.join('\n')}\n });`; + return { up, down }; +} - const down = isEmpty - ? ` // No changes needed` - : ` await knex.schema.alterTable('${tableName}', (table) => {\n${downLines.join('\n')}\n });`; +/** @internal PostgreSQL Unicode literals avoid Knex parsing literal `?` as a binding. */ +function defaultStringLiteral(value: string): string { + return `U&'${value.replaceAll('\\', '\\\\').replaceAll("'", "''").replaceAll('?', '\\003f')}'`; +} - return { up, down }; +/** @internal Generate default-only DDL without altering column types or nullability. */ +function defaultChangeCode(table: string, column: string, value: any): string { + const names = `${JSON.stringify(table)}, ${JSON.stringify(column)}`; + if (value === null || value === undefined) { + return ` await knex.raw('ALTER TABLE ?? ALTER COLUMN ?? DROP DEFAULT', [${names}]);`; + } + const expression = + value === 'now' + ? 'knex.fn.now()' + : typeof value === 'string' + ? `knex.raw(${JSON.stringify(defaultStringLiteral(value))})` + : typeof value === 'object' && value !== null && value.raw + ? `knex.raw(${JSON.stringify(value.raw)})` + : JSON.stringify(value); + return ` await knex.raw(knex.raw('ALTER TABLE ?? ALTER COLUMN ?? SET DEFAULT ?', [${names}, ${expression}]).toQuery().replaceAll('?', '\\\\?'));`; } /** @internal Build a Knex ColumnBuilder from a PostgreSQL data type string (runtime use). */ diff --git a/libs/knex-schema/src/operations/insert.test.ts b/libs/knex-schema/src/operations/insert.test.ts new file mode 100644 index 00000000..32095d5b --- /dev/null +++ b/libs/knex-schema/src/operations/insert.test.ts @@ -0,0 +1,139 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { mockDriver } from '../../testing/mock-driver.js'; +import { number, object, query, string } from '../index.js'; + +const Item = object({ + id: number().primaryKey(), + name: string().hasColumnName('display_name'), + version: number() +}).hasTableName('items'); +const connections: ReturnType[] = []; +function setup(schema = Item) { + const driver = mockDriver(); + connections.push(driver); + driver.respond = q => + q.sql.startsWith('insert') + ? [{ id: 1, display_name: 'saved', version: 2 }] + : []; + return { driver, writer: query(schema).query(driver.knex) }; +} +afterEach(async () => { + await Promise.all(connections.splice(0).map(d => d.knex.destroy())); +}); + +describe('insert execution', () => { + it('maps columns, executes ordered hooks and does not mutate inputs', async () => { + const before = vi.fn(data => ({ ...data, name: data.name.trim() })); + const after = vi.fn(); + const schema = Item.hasTimestamps() + .beforeInsert(before) + .beforeInsert(() => undefined) + .afterInsert(after); + const { driver, writer } = setup(schema); + const input = { id: 1, name: ' submitted ', version: 2 }; + const row = await writer.insert(input); + expect(row).toEqual({ id: 1, name: 'saved', version: 2 }); + expect(input.name).toBe(' submitted '); + expect(driver.queries[0].sql).toContain('CURRENT_TIMESTAMP'); + expect(driver.queries[0].bindings).toContain('submitted'); + expect(after).toHaveBeenCalledWith(row); + expect(after.mock.invocationCallOrder[0]).toBeGreaterThan( + before.mock.invocationCallOrder[0] + ); + await writer.insertMany([input, { ...input, id: 2 }]); + expect(before).toHaveBeenCalledTimes(3); + }); + + it('supports ignored conflicts, implicit merges, and conditional expression updates', async () => { + const { driver, writer } = setup(Item.hasTimestamps()); + await writer.onConflict('id').merge({ id: 1, name: 'a', version: 1 }); + const merge = driver.queries[0].sql.split('do update set')[1]; + expect(merge).toContain('updated_at'); + expect(merge).not.toContain('created_at'); + await writer.onConflict('id').merge( + { id: 1, name: 'a', version: 1 }, + { + version: h => h.raw('?? + ?', [h.column('version'), 1]), + name: h => h.excluded('name') + }, + { + where: (qb, h) => + qb.where(h.column('version'), '<', h.excluded('version')) + } + ); + expect(driver.queries[1].sql).toContain('excluded."display_name"'); + expect(driver.queries[1].bindings).toContain(1); + expect(driver.queries[1].sql).toContain( + 'where "version" < excluded."version"' + ); + await writer + .onConflict('id') + .merge( + { id: 1, name: 'a', version: 1 }, + { where: qb => qb.where('version', 1) } + ); + expect(driver.queries[2].sql).toContain('where "version" ='); + driver.respond = () => []; + expect( + await writer + .onConflict('id') + .ignore({ id: 1, name: 'a', version: 1 }) + ).toBeUndefined(); + expect(driver.queries[3].sql).toContain('do nothing'); + }); + + it('upserts either all columns or only the selected update columns', async () => { + const { driver, writer } = setup(); + await writer.upsert( + { id: 1, name: 'a', version: 1 }, + { conflictColumns: ['id'], updateColumns: ['name'] } + ); + expect( + driver.queries[0].sql + .split('do update set')[1] + .split(' returning')[0] + ).toBe(' "display_name" = excluded."display_name"'); + await writer.upsert( + { id: 1, name: 'b', version: 2 }, + { conflictColumns: ['id'] } + ); + expect(driver.queries[1].sql.split('do update set')[1]).toContain( + '"version" = excluded."version"' + ); + }); + + it('chunks bulk writes, runs hooks, and never updates conflict or creation columns', async () => { + const before = vi.fn(data => data); + const after = vi.fn(); + const { driver, writer } = setup( + Item.hasTimestamps().beforeInsert(before).afterInsert(after) + ); + const rows = [1, 2, 3].map(id => ({ id, name: 'n', version: 1 })); + const result = await writer.bulkInsert(rows, { + chunkSize: 2, + onConflict: 'merge', + conflictColumns: ['id'] + }); + expect(driver.queries).toHaveLength(2); + expect(before).toHaveBeenCalledTimes(3); + expect(after).toHaveBeenCalledTimes(2); + expect(result).toHaveLength(2); + const merge = driver.queries[0].sql.split('do update set')[1]; + expect(merge).not.toContain('"id" ='); + expect(merge).not.toContain('"created_at" ='); + expect(merge).toContain('"updated_at" ='); + await writer.bulkInsert(rows, { + chunkSize: 0, + onConflict: 'ignore', + conflictColumns: ['id'] + }); + expect(driver.queries.slice(2)).toHaveLength(3); + expect(driver.queries[2].sql).toContain('do nothing'); + await writer.bulkUpsert(rows, { conflictColumns: ['id'] }); + expect(driver.queries.at(-1)?.sql).toContain('do update set'); + expect(await writer.bulkInsert([])).toEqual([]); + await expect( + writer.bulkInsert(rows, { onConflict: 'merge' }) + ).rejects.toThrow('conflictColumns'); + }); +}); diff --git a/libs/knex-schema/src/operations/pagination.test.ts b/libs/knex-schema/src/operations/pagination.test.ts new file mode 100644 index 00000000..247dca74 --- /dev/null +++ b/libs/knex-schema/src/operations/pagination.test.ts @@ -0,0 +1,183 @@ +import { afterEach, describe, expect, it } from 'vitest'; +import { mockDriver } from '../../testing/mock-driver.js'; +import { + aggregate, + boolean, + date, + number, + object, + query, + string +} from '../index.js'; +import { QuerySource } from '../QuerySource.js'; + +const Row = object({ + id: number().primaryKey(), + name: string(), + active: boolean(), + created: date(), + optional: string().optional() +}).hasTableName('items'); +const drivers: ReturnType[] = []; +function setup() { + const driver = mockDriver(); + drivers.push(driver); + driver.respond = q => + q.sql.includes('count(') + ? [{ count: '5' }] + : [1, 2, 3].map(id => ({ + id, + name: `n${id}`, + active: true, + created: new Date(0), + optional: null, + __cursor_value: id, + ...Object.fromEntries( + [...q.sql.matchAll(/as "(__cb_cursor_\d+)"/g)].map( + ([_, key], i) => [key, String(i ? id : id)] + ) + ) + })); + return { + driver, + read: query(Row).query(driver.knex), + source: () => new QuerySource(driver.knex, Row) + }; +} +afterEach(async () => { + await Promise.all(drivers.splice(0).map(d => d.knex.destroy())); +}); +describe('pagination execution and cursor validation', () => { + it('counts and pages without modifying immutable reads', async () => { + const { read, driver } = setup(); + const sql = read.toQuery(); + const result = await read.paginate({ page: 2, pageSize: 2 }); + expect(result).toMatchObject({ + total: 5, + totalPages: 3, + hasNextPage: true, + hasPreviousPage: true + }); + expect(read.toQuery()).toEqual(sql); + expect(driver.queries[1].bindings).toEqual([2, 2]); + expect(await read.pluck('id')).toEqual([1, 2, 3]); + for (const options of [ + { page: 0, pageSize: 2 }, + { page: 1, pageSize: 0 } + ]) + await expect(read.paginate(options)).rejects.toThrow('positive'); + driver.respond = () => [{ count: '9007199254740993' }]; + await expect(read.paginate({ page: 1, pageSize: 2 })).rejects.toThrow( + 'safe integer' + ); + }); + it('returns raw and composite cursors and rejects incompatible/tampered tokens', async () => { + const { read, driver, source } = setup(); + const page = await read.paginateAfter({ + limit: 2, + cursor: 4, + direction: 'asc' + }); + expect(page).toMatchObject({ hasMore: true, nextCursor: '2' }); + expect(driver.queries[0].bindings).toContain(4); + const orderBy = [{ column: 'id', direction: 'asc' }] as const; + const first = await source() + .where('active', true) + .paginateAfter({ limit: 2, orderBy }); + expect(first.data).toHaveLength(2); + expect(first.data[0]).not.toHaveProperty('__cb_cursor_0'); + expect(first.nextCursor).toBeTypeOf('string'); + await source() + .where('active', true) + .paginateAfter({ limit: 2, orderBy, cursor: first.nextCursor }); + expect(driver.queries.at(-1)?.bindings).toContain('2'); + const payload = JSON.parse( + Buffer.from(first.nextCursor!, 'base64url').toString() + ); + for (const cursor of [ + 'invalid!', + Buffer.from(JSON.stringify({ ...payload, v: 2 })).toString( + 'base64url' + ), + Buffer.from( + JSON.stringify({ ...payload, values: ['not a number'] }) + ).toString('base64url') + ]) + await expect( + source().paginateAfter({ limit: 2, orderBy, cursor }) + ).rejects.toThrow('Invalid cursor'); + await expect( + source().paginateAfter({ + limit: 2, + orderBy: [{ column: 'id', direction: 'desc' }], + cursor: first.nextCursor + }) + ).rejects.toThrow('incompatible'); + driver.respond = () => []; + expect(await read.paginateAfter({ limit: 2 })).toEqual({ + data: [], + hasMore: false, + nextCursor: null + }); + }); + it('validates cursor shapes and unsafe query combinations before executing', async () => { + const { read, source, driver } = setup(); + for (const options of [ + { limit: 0 }, + { limit: 1, direction: 'wrong' }, + { limit: 1, column: 'missing' } + ]) + await expect( + (read as any).paginateAfter(options) + ).rejects.toThrow(); + for (const options of [ + { limit: 0, orderBy: [{ column: 'id', direction: 'asc' }] }, + { limit: 1, orderBy: [] }, + { limit: 1, orderBy: [{ column: 'id', direction: 'wrong' }] }, + { limit: 1, orderBy: [{ column: 'optional', direction: 'asc' }] }, + { limit: 1, orderBy: [{ column: 'name', direction: 'asc' }] }, + { + limit: 1, + orderBy: [ + { column: 'id', direction: 'asc' }, + { column: 'id', direction: 'desc' } + ] + } + ]) + await expect( + source().paginateAfter(options as any) + ).rejects.toThrow(); + await expect( + source() + .offset(1) + .paginateAfter({ + limit: 2, + orderBy: [{ column: 'id', direction: 'asc' }] + }) + ).rejects.toThrow('offsets'); + await expect( + read + .select(() => ({ count: aggregate.count() })) + .paginateAfter({ limit: 1 }) + ).rejects.toThrow('aggregate'); + expect(driver.queries).toEqual([]); + }); + it('executes low-level offset and simple cursor pages using mapped fields', async () => { + const { source } = setup(); + expect(await source().paginate({ page: 1, pageSize: 2 })).toMatchObject( + { total: 5, totalPages: 3 } + ); + expect( + await source().paginateAfter({ + limit: 2, + column: t => t.id, + direction: 'asc', + cursor: 0 + }) + ).toMatchObject({ hasMore: true, nextCursor: '2' }); + expect(await source().paginateAfter({ limit: 5 })).toMatchObject({ + hasMore: false, + nextCursor: null + }); + }); +}); diff --git a/libs/knex-schema/src/operations/relations.test.ts b/libs/knex-schema/src/operations/relations.test.ts new file mode 100644 index 00000000..48fd2ebe --- /dev/null +++ b/libs/knex-schema/src/operations/relations.test.ts @@ -0,0 +1,195 @@ +import { afterEach, describe, expect, it } from 'vitest'; +import { mockDriver } from '../../testing/mock-driver.js'; +import { array, defineEntity, number, object, string } from '../index.js'; +import { QuerySource } from '../QuerySource.js'; + +const Owner = object({ + id: number().primaryKey(), + name: string().hasColumnName('display_name'), + rootId: number().optional() +}).hasTableName('owners'); +const Child = object({ + id: number().primaryKey(), + rootId: number(), + title: string() +}).hasTableName('children'); +const Base = object({ + id: number().primaryKey(), + kind: string(), + ownerId: number(), + owner: Owner.optional(), + single: Child.optional(), + children: array(Child).optional(), + tags: array(Owner).optional() +}).hasTableName('roots'); +const Root = defineEntity(Base) + .belongsTo( + t => t.owner, + t => t.ownerId, + t => t.id + ) + .hasOne( + t => t.single, + t => t.id, + t => t.rootId + ) + .hasMany( + t => t.children, + t => t.id, + t => t.rootId + ) + .belongsToMany(t => t.tags, { + table: 'root_tags', + localKey: 'root_id', + foreignKey: 'owner_id' + }); +const Body = defineEntity( + object({ + rootId: number().primaryKey(), + caption: string().hasColumnName('caption_text'), + ownerId: number(), + owner: Owner.optional(), + single: Child.optional() + }).hasTableName('details') +) + .belongsTo( + t => t.owner, + t => t.ownerId, + t => t.id + ) + .hasOne( + t => t.single, + t => t.rootId, + t => t.rootId + ); +const Poly = defineEntity( + object({ id: number().primaryKey(), kind: string() }).hasTableName('roots') +) + .discriminator('kind') + .ctiVariant('photo', Body, t => t.rootId) + .stiVariant('note', object({ text: string() })); +const drivers: ReturnType[] = []; +function setup(schema: any) { + const driver = mockDriver(); + drivers.push(driver); + return { driver, source: () => new QuerySource(driver.knex, schema) }; +} +afterEach(async () => { + await Promise.all(drivers.splice(0).map(d => d.knex.destroy())); +}); + +describe('relation SQL planner', () => { + it.each(['owner', 'single', 'children', 'tags'])( + 'plans %s relations with and without customization', + async name => { + const { driver, source } = setup(Root.schema); + for (const customize of [ + undefined, + (q: any) => q.select('id').orderBy('id').limit(2).offset(1) + ]) { + const plan = source().include(name, customize); + const sql = plan.toQuery(); + expect(sql).toContain('join'); + expect(sql).toContain(name); + if (name === 'tags') expect(sql).toContain('root_tags'); + if (name === 'children') expect(sql).toContain('jsonb_agg'); + driver.respond = () => [ + { + id: 1, + kind: 'plain', + ownerId: 2, + [name]: + name === 'children' || name === 'tags' ? [] : null + } + ]; + expect((await plan.execute())[0][name]).toEqual( + name === 'children' || name === 'tags' ? [] : null + ); + } + expect(() => source().include('missing')).toThrow( + 'Unknown relation' + ); + } + ); + it('maps CTI bodies and relations, including absent optional rows', async () => { + const { driver, source } = setup(Poly.schema); + const plan = source() + .includeVariant('photo', 'owner', q => { + q.select('name'); + }) + .includeVariant('photo', 'single'); + const sql = plan.toQuery(); + expect(sql).toContain('__v_photo'); + expect(sql).toContain('caption_text'); + driver.respond = () => [ + { + id: 1, + kind: 'photo', + __v_photo__rootId: 1, + __v_photo__caption_text: 'caption', + __v_photo__ownerId: 2, + __v_photo__rel_owner__display_name: 'Ada', + __v_photo__rel_single__id: null + } + ]; + expect(await plan.execute()).toMatchObject([ + { + id: 1, + kind: 'photo', + caption: 'caption', + ownerId: 2, + owner: { name: 'Ada' }, + single: null + } + ]); + driver.respond = () => [ + { id: 1, kind: 'photo', __v_photo__rootId: null } + ]; + await expect(source().execute()).rejects.toThrow('Polymorphic orphan'); + driver.respond = () => [ + { id: 2, kind: 'note', text: 'hello' }, + { id: 3, kind: 'unknown' } + ]; + expect(await source().execute()).toEqual([ + { id: 2, kind: 'note', text: 'hello' }, + { id: 3, kind: 'unknown' } + ]); + }); + it('resolves unique variant relation names and validates variant filters', () => { + const { source } = setup(Poly.schema); + expect(source().include('owner').toQuery()).toContain( + '__v_photo__rel_owner' + ); + expect( + source() + .whereVariant('photo', 'caption', 'LIKE', '%safe%') + .selectVariants(['photo']) + .toQuery() + ).toContain('%safe%'); + expect( + source().whereVariant('note', 'text', '=', 'hello').toQuery() + ).toContain('hello'); + expect(() => source().whereVariant('missing', 'text', '=', 1)).toThrow( + 'unknown variant' + ); + expect(() => + source().whereVariant('note', 'text', 'invalid', 1) + ).toThrow('not allowed'); + expect(() => source().includeVariant('missing', 'owner')).toThrow( + 'unknown variant' + ); + expect(() => source().includeVariant('note', 'missing')).toThrow( + 'unknown relation' + ); + const normal = setup(Base).source; + expect(() => normal().includeVariant('photo', 'owner')).toThrow( + 'not polymorphic' + ); + expect(() => normal().whereVariant('photo', 'caption', '=', 1)).toThrow( + 'polymorphic schema' + ); + expect(() => normal().selectVariants(['photo'])).toThrow( + 'polymorphic schema' + ); + }); +}); diff --git a/libs/knex-schema/src/operations/select.test.ts b/libs/knex-schema/src/operations/select.test.ts new file mode 100644 index 00000000..52f766d6 --- /dev/null +++ b/libs/knex-schema/src/operations/select.test.ts @@ -0,0 +1,71 @@ +import { afterEach, describe, expect, it } from 'vitest'; +import { mockDriver } from '../../testing/mock-driver.js'; +import { aggregate, number, object, string } from '../index.js'; +import { QuerySource } from '../QuerySource.js'; + +const Schema = object({ + id: number().primaryKey(), + name: string().hasColumnName('display_name') +}).hasTableName('items'); +const drivers: ReturnType[] = []; +function setup(schema: any = Schema) { + const driver = mockDriver(); + drivers.push(driver); + return { driver, read: () => new QuerySource(driver.knex, schema) }; +} +afterEach(async () => { + await Promise.all(drivers.splice(0).map(d => d.knex.destroy())); +}); +describe('native selection planning', () => { + it.each(['count', 'countDistinct', 'min', 'max', 'sum', 'avg'])( + 'maps %s columns and compiles aggregate SQL', + async method => { + const { read } = setup(); + expect((read() as any)[method]('id').toQuery()).toContain( + method === 'countDistinct' ? 'count(distinct' : `${method}(` + ); + if (method.startsWith('count')) + expect((read() as any)[method]().toQuery()).toContain('count('); + } + ); + it('compiles grouped/having projections and rejects invalid projection values', async () => { + const { read, driver } = setup(); + const q = read() + .select(t => ({ label: t.name, count: aggregate.count() })) + .groupBy('name') + .having('id', '>', 0); + expect(q.toQuery()).toContain('group by "display_name"'); + driver.respond = () => [{ label: 'group', count: '2' }]; + expect(await q.execute()).toEqual([{ label: 'group', count: 2 }]); + expect(() => + read().select(() => ({ invalid: 'not a descriptor' }) as any) + ).toThrow('property descriptor'); + expect(() => + read() + .select(t => ({ label: t.name })) + .select('id') + ).toThrow(); + expect(read().distinct('name').toQuery()).toContain( + 'distinct "display_name"' + ); + expect(read().selectRaw('? as literal', [1]).toQuery()).toContain( + '1 as literal' + ); + expect(read().selectRaw('1 as literal').toQuery()).toContain( + '1 as literal' + ); + }); + it('applies explicit scopes and isolates default visibility', () => { + const schema = Schema.defaultScope(q => q.where('id', '>', 0)) + .scope('named', q => q.where('name', 'visible')) + .softDelete(); + const { read } = setup(schema); + expect(read().scoped('named').toQuery()).toContain("'visible'"); + expect(read().toQuery()).toContain('"deleted_at" is null'); + expect(read().onlyDeleted().toQuery()).toContain( + '"deleted_at" is not null' + ); + expect(read().unscoped().toQuery()).not.toContain('deleted_at'); + expect(() => read().scoped('missing' as any)).toThrow('Unknown scope'); + }); +}); diff --git a/libs/knex-schema/src/polymorphic-execution.test.ts b/libs/knex-schema/src/polymorphic-execution.test.ts new file mode 100644 index 00000000..a7650310 --- /dev/null +++ b/libs/knex-schema/src/polymorphic-execution.test.ts @@ -0,0 +1,140 @@ +import { afterEach, describe, expect, it } from 'vitest'; +import { mockDriver } from '../testing/mock-driver.js'; +import { + boolean, + date, + defineEntity, + number, + object, + query, + string +} from './index.js'; + +const Base = object({ + id: number().primaryKey(), + kind: string(), + enabled: boolean(), + created: date(), + deletedAt: date().optional().hasColumnName('deleted_at') +}) + .hasTableName('assets') + .softDelete(); +const Entity = defineEntity(Base) + .discriminator('kind') + .stiVariant('note', object({ text: string() })) + .stiVariant('image', object({ url: string() })); +const drivers: ReturnType[] = []; +function setup(schema: any = Entity.schema) { + const driver = mockDriver(); + drivers.push(driver); + const row = { + id: 1, + kind: 'note', + enabled: true, + created: new Date(0).toISOString(), + deletedAt: null, + text: 'hello' + }; + driver.respond = q => + q.sql.includes('count(') ? [{ count: '3' }] : [{ __read_poly: row }]; + return { driver, read: query(schema).query(driver.knex), row }; +} +afterEach(async () => { + await Promise.all(drivers.splice(0).map(d => d.knex.destroy())); +}); +describe('polymorphic read execution', () => { + it('globally orders, pages, plucks and decodes branch rows without mutating the definition', async () => { + const { read } = setup(); + const original = read.toQuery(); + const page = read + .selectVariants(['note']) + .orderBy('id', 'desc') + .orderBy('enabled') + .orderBy('created') + .orderBy('kind') + .orderByRaw('1 desc') + .limit(2) + .offset(1); + expect(await page.first()).toMatchObject({ + kind: 'note', + text: 'hello', + created: new Date(0) + }); + expect(await page.pluck('id')).toEqual([1]); + expect(await page.countValue()).toBe(3); + expect(await page.paginate({ page: 2, pageSize: 2 })).toMatchObject({ + total: 3, + totalPages: 2, + hasNextPage: false, + hasPreviousPage: true + }); + expect(read.toQuery()).toBe(original); + expect(read.onlyDeleted().toQuery()).toContain('is not null'); + expect(read.withDeleted().toQuery()).not.toMatch( + /"__cb_read_deleted_\d+" is null/ + ); + }); + it('validates bounds, projections, selected branches and decoded discriminators', async () => { + const { read, driver, row } = setup(); + for (const method of ['limit', 'offset']) + expect(() => read[method](-1)).toThrow('non-negative'); + for (const keys of [[], ['missing'], ['note', 'note']]) + expect(() => read.selectVariants(keys)).toThrow( + 'declared variants' + ); + expect(() => read.orderBy('id', 'wrong')).toThrow('direction'); + expect(() => read.orderBy('missing')).toThrow('Column'); + expect(() => read.forVariant('missing', (q: any) => q)).toThrow(); + expect(() => + read.forVariant('note', (q: any) => q.select('text')) + ).toThrow('discriminator'); + expect(() => read.include('missing')).toThrow('Unknown relation'); + expect(() => read.scoped('missing')).toThrow('Unknown scope'); + for (const opts of [ + { page: 0, pageSize: 1 }, + { page: 1, pageSize: 0 } + ]) + await expect(read.paginate(opts)).rejects.toThrow('positive'); + driver.respond = () => [{ count: '9007199254740993' }]; + await expect(read.countValue()).rejects.toThrow('safe integer'); + driver.respond = () => [{ __read_poly: { ...row, kind: 'missing' } }]; + await expect(read.execute()).rejects.toThrow('unknown polymorphic'); + }); + it('applies stable named/default scopes and rejects shape-changing scopes', () => { + const schema = Entity.schema + .defaultScope(q => + q.where('enabled', true).orderBy('id').limit(10).offset(1) + ) + .scope('notes', q => q.where('kind', 'note')); + const { read } = setup(schema); + expect(read.scoped('notes').toQuery()).toContain("'note'"); + expect(read.unscoped().toQuery()).not.toContain('limit 10'); + const invalid = Entity.schema.scope('invalid', (q: any) => + q.selectVariants(['note']) + ); + expect(() => setup(invalid).read.scoped('invalid')).toThrow( + 'shape-preserving' + ); + }); + it('requires explicit output contracts for raw query escapes', async () => { + const { read, driver } = setup(); + const output = object({ id: number() }); + driver.respond = () => [{ id: 1 }]; + expect( + await read.selectRaw('? as id', [1], { output }).execute() + ).toEqual([{ id: 1 }]); + expect( + await read + .apply( + (q: any) => { + q.clearSelect().select('id'); + }, + { output } + ) + .first() + ).toEqual({ id: 1 }); + expect(() => read.apply(() => Promise.resolve(), { output })).toThrow( + 'synchronously' + ); + }); +}); diff --git a/libs/knex-schema/testing/mock-driver.ts b/libs/knex-schema/testing/mock-driver.ts new file mode 100644 index 00000000..cfda45a9 --- /dev/null +++ b/libs/knex-schema/testing/mock-driver.ts @@ -0,0 +1,45 @@ +import Knex from 'knex'; + +export type CapturedQuery = { + sql: string; + bindings: unknown[]; + method: string; +}; + +/** Exercise real Knex compilation while replacing only the database driver. */ +export function mockDriver() { + const knex = Knex({ client: 'pg' }); + const queries: CapturedQuery[] = []; + const fixture = { + knex, + queries, + respond: (_query: CapturedQuery): unknown[] => [] + }; + knex.client.acquireConnection = async () => ({ + query(options: any, callback: any) { + const captured = { + sql: options.text ?? options, + bindings: options.values ?? [], + method: String(options.text ?? options) + .split(' ')[0] + .toLowerCase() + }; + queries.push(captured); + try { + const rows = fixture.respond(captured); + callback(null, { + rows, + rowCount: rows.length, + command: + captured.method === 'with' + ? 'SELECT' + : captured.method.toUpperCase() + }); + } catch (error) { + callback(error); + } + } + }); + knex.client.releaseConnection = async () => {}; + return fixture; +} diff --git a/libs/mapper/README.md b/libs/mapper/README.md index 32bfc361..877ad014 100644 --- a/libs/mapper/README.md +++ b/libs/mapper/README.md @@ -3,7 +3,7 @@ [![CI](https://github.com/cleverbrush/framework/actions/workflows/ci.yml/badge.svg)](https://github.com/cleverbrush/framework/actions/workflows/ci.yml) [![License: BSD-3-Clause](https://img.shields.io/badge/license-BSD--3--Clause-blue.svg)](../../LICENSE) -![Unit coverage](https://img.shields.io/badge/unit_coverage-93.9%25-brightgreen) +![Unit coverage](https://img.shields.io/badge/unit_coverage-98.8%25-brightgreen) A type-safe, declarative object mapper for converting objects between different `@cleverbrush/schema` representations. Uses PropertyDescriptors as pointers to properties (similar to expressions in C# .NET) and enforces **compile-time completeness** — TypeScript will produce an error if any target property is not mapped, auto-mapped, or explicitly ignored. diff --git a/libs/mapper/src/MappingRegistry.arrays.test.ts b/libs/mapper/src/MappingRegistry.arrays.test.ts new file mode 100644 index 00000000..c7f05e7a --- /dev/null +++ b/libs/mapper/src/MappingRegistry.arrays.test.ts @@ -0,0 +1,196 @@ +import { array, number, object, string } from '@cleverbrush/schema'; +import { expect, it } from 'vitest'; +import { Mapper, MappingRegistry } from './MappingRegistry.js'; + +it('maps nested primitive arrays synchronously and asynchronously', async () => { + const schema = object({ rows: array(array(number())).optional() }); + const mapping = new Mapper(schema, schema) + .for(t => t.rows) + .from(s => s.rows); + const asyncMap = mapping.getMapper(); + const syncMap = mapping.getSyncMapper(); + for (const input of [{}, { rows: null }, { rows: [[1, 2], [], null, 7] }]) { + expect(await asyncMap(input as any)).toEqual(syncMap(input as any)); + } + expect(syncMap({ rows: [[1, 2], []] })).toEqual({ rows: [[1, 2], []] }); + await expect(asyncMap({ rows: 'invalid' } as any)).rejects.toThrow( + 'Expected array' + ); + expect(() => syncMap({ rows: 'invalid' } as any)).toThrow('Expected array'); +}); + +it('skips missing nested source descriptors in both executors', async () => { + const source = object({ child: object({ name: string() }) }); + const target = object({ name: string() }); + const mapping = new Mapper(source, target) + .for(t => t.name) + .from(s => s.child.name); + expect(await mapping.getMapper()({} as any)).toEqual({}); + expect(mapping.getSyncMapper()({} as any)).toEqual({}); +}); + +it('ignores explicitly excluded target fields in both executors', async () => { + const source = object({ id: number() }); + const target = object({ id: number(), ignored: string().optional() }); + const mapping = new Mapper(source, target) + .for(t => t.id) + .from(s => s.id) + .for(t => t.ignored) + .ignore(); + expect(await mapping.getMapper()({ id: 1 })).toEqual({ id: 1 }); + expect(mapping.getSyncMapper()({ id: 1 })).toEqual({ id: 1 }); +}); + +it('copies object arrays without requiring a registry', async () => { + const schema = object({ items: array(object({ id: number() })) }); + const mapping = new Mapper(schema, schema) + .for(t => t.items) + .from(s => s.items); + const input = { items: [{ id: 1 }, { id: 2 }] }; + expect(await mapping.getMapper()(input)).toEqual(input); + expect(mapping.getSyncMapper()(input)).toEqual(input); +}); + +it('copies assignable arrays without projecting away extra source fields', async () => { + const from = object({ + items: array(object({ id: number(), name: string() })) + }); + const to = object({ items: array(object({ id: number() })) }); + const mapping = new Mapper(from, to).for(t => t.items).from(s => s.items); + const input = { items: [{ id: 1, name: 'Example' }] }; + expect(await mapping.getMapper()(input)).toEqual(input); + expect(mapping.getSyncMapper()(input)).toEqual(input); +}); + +it('awaits asynchronous element mappings through nested arrays', async () => { + const item = object({ id: number() }); + const dto = object({ label: string() }); + const from = object({ items: array(array(item)) }); + const to = object({ items: array(array(dto)) }); + const registry = new MappingRegistry() + .configure(item, dto, mapper => + mapper.for(t => t.label).compute(async s => String(s.id)) + ) + .configure(from, to, mapper => mapper); + expect( + await registry.getMapper(from, to)({ items: [[{ id: 2 }], []] }) + ).toEqual({ items: [[{ label: '2' }], []] }); + expect(() => (registry as any).getSyncMapper(from, to)).toThrow(); +}); + +it('skips unavailable nested array paths in both executors', async () => { + const from = object({ parent: object({ items: array(number()) }) }); + const to = object({ items: array(number()).optional() }); + const mapping = new Mapper(from, to) + .for(t => t.items) + .from(s => s.parent.items); + expect(await mapping.getMapper()({} as any)).toEqual({}); + expect(mapping.getSyncMapper()({} as any)).toEqual({}); +}); + +it('rejects primitive/object mixed array element types instead of copying them', () => { + const from = object({ items: array(number()) }); + const to = object({ items: array(object({ id: number() })) }); + expect(() => + (new MappingRegistry() as any).configure(from, to, (m: any) => m) + ).toThrow(/not mapped/); +}); + +it('rejects asymmetric object, primitive and nested array element shapes', () => { + for (const [left, right] of [ + [object({ id: number() }), number()], + [array(number()), number()], + [number(), array(number())], + [object({ id: number(), name: string() }), object({ id: number() })] + ]) { + const from = object({ items: array(left) }); + const to = object({ items: array(right) }); + expect(() => + (new MappingRegistry() as any).configure(from, to, (m: any) => m) + ).toThrow(/not mapped/); + } +}); + +it('requires explicit mappings for incompatible nested object shapes', () => { + const from = object({ child: object({ id: number(), name: string() }) }); + for (const child of [ + object({ id: number() }), + object({ id: number(), label: string() }) + ]) { + const to = object({ child }); + expect(() => + (new MappingRegistry() as any).configure(from, to, (m: any) => m) + ).toThrow(/not mapped/); + } +}); + +it('does not configure incompatible array elements automatically', () => { + const source = object({ rows: array(array(number())) }); + const target = object({ rows: array(array(string())) }); + const registry = new MappingRegistry(); + expect(() => + (registry as any).configure(source, target, (mapper: any) => mapper) + ).toThrow(/not mapped/); + const objectSource = object({ rows: array(object({ id: number() })) }); + const objectTarget = object({ rows: array(object({ name: string() })) }); + expect(() => + (registry as any).configure( + objectSource, + objectTarget, + (mapper: any) => mapper + ) + ).toThrow(/not mapped/); +}); + +it('copies equivalent object array elements and preserves missing optional nested objects', async () => { + const item = object({ id: number() }); + const schema = object({ rows: array(item), child: item.optional() }); + const registry = new MappingRegistry().configure( + schema, + schema, + mapper => mapper + ); + const syncMap = registry.getSyncMapper(schema, schema); + expect(syncMap({ rows: [{ id: 1 }] })).toEqual({ rows: [{ id: 1 }] }); + expect( + await registry.getMapper(schema, schema)({ rows: [{ id: 1 }] }) + ).toEqual({ rows: [{ id: 1 }] }); + expect(syncMap({ rows: [{ id: 2 }], child: { id: 3 } })).toEqual({ + rows: [{ id: 2 }], + child: { id: 3 } + }); +}); + +it('maps nested registered arrays and skips unavailable optional source paths', async () => { + const sourceItem = object({ id: number() }); + const targetItem = object({ label: string() }); + const source = object({ + child: sourceItem, + rows: array(array(sourceItem)) + }); + const target = object({ + child: targetItem, + rows: array(array(targetItem)) + }); + const registry = new MappingRegistry() + .configure(sourceItem, targetItem, mapper => + mapper.for(t => t.label).compute(s => String(s.id)) + ) + .configure(source, target, mapper => mapper); + const syncMap = registry.getSyncMapper(source, target); + expect(syncMap({ child: undefined, rows: [] } as any)).toEqual({ + rows: [] + }); + expect( + await registry.getMapper( + source, + target + )({ child: undefined, rows: [] } as any) + ).toEqual({ rows: [] }); + expect(syncMap({ rows: [[{ id: 2 }]] } as any)).toEqual({ + rows: [[{ label: '2' }]] + }); + expect( + await registry.getMapper(source, target)({ rows: [[{ id: 2 }]] } as any) + ).toEqual({ rows: [[{ label: '2' }]] }); +}); diff --git a/libs/orm-cli/README.md b/libs/orm-cli/README.md index 929da28b..c2dd885d 100644 --- a/libs/orm-cli/README.md +++ b/libs/orm-cli/README.md @@ -1,6 +1,6 @@ # `@cleverbrush/orm-cli` -![Unit coverage](https://img.shields.io/badge/unit_coverage-61.2%25-yellow) +![Unit coverage](https://img.shields.io/badge/unit_coverage-98.7%25-brightgreen) A standalone CLI for managing PostgreSQL schema migrations for projects built diff --git a/libs/orm-cli/src/cli-routing.test.ts b/libs/orm-cli/src/cli-routing.test.ts new file mode 100644 index 00000000..a37d937a --- /dev/null +++ b/libs/orm-cli/src/cli-routing.test.ts @@ -0,0 +1,126 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { run } from './cli.js'; + +const commands = vi.hoisted(() => ({ + load: vi.fn(), + generate: vi.fn(), + migrate: vi.fn(), + rollback: vi.fn(), + status: vi.fn(), + push: vi.fn(), + validate: vi.fn() +})); +vi.mock('./config.js', () => ({ loadConfig: commands.load })); +vi.mock('./commands/generate.js', () => ({ generate: commands.generate })); +vi.mock('./commands/run.js', () => ({ run: commands.migrate })); +vi.mock('./commands/rollback.js', () => ({ rollback: commands.rollback })); +vi.mock('./commands/status.js', () => ({ status: commands.status })); +vi.mock('./commands/push.js', () => ({ push: commands.push })); +vi.mock('./commands/validate.js', () => ({ validate: commands.validate })); + +describe('CLI routing and connection cleanup', () => { + const destroy = vi.fn(); + const config = { knex: { destroy } }; + beforeEach(() => { + vi.resetAllMocks(); + commands.load.mockResolvedValue(config); + destroy.mockResolvedValue(undefined); + vi.spyOn(console, 'log').mockImplementation(() => {}); + vi.spyOn(console, 'error').mockImplementation(() => {}); + vi.spyOn(process, 'exit').mockImplementation(() => { + throw new Error('process exited'); + }); + }); + afterEach(() => vi.restoreAllMocks()); + + it.each([[], ['--help'], ['-h'], ['--version'], ['-v']])( + 'prints informational output without opening the database: %j', + async (...args) => { + await run(args); + expect(console.log).toHaveBeenCalled(); + expect(commands.load).not.toHaveBeenCalled(); + expect(destroy).not.toHaveBeenCalled(); + } + ); + + it.each([ + ['run', 'migrate', '--to', 'initial.ts'], + ['rollback', 'rollback', '--all'], + ['status', 'status', '--dir', 'migrations'] + ])( + 'routes migrate %s and destroys the pool', + async (sub, key, ...flags) => { + await run(['migrate', sub, ...flags, '--config', 'database.ts']); + expect(commands.load).toHaveBeenCalledWith('database.ts'); + expect(commands[key]).toHaveBeenCalledWith( + config, + expect.objectContaining({ '--config': 'database.ts' }) + ); + expect(destroy).toHaveBeenCalledOnce(); + } + ); + + it('skips flags when choosing migration names and supplies the default', async () => { + await run([ + 'migrate', + 'generate', + '--dir', + 'tmp', + '--yes', + '--config', + 'db.ts', + 'add_users' + ]); + expect(commands.generate).toHaveBeenLastCalledWith( + 'add_users', + config, + { '--dir': 'tmp', '--yes': true, '--config': 'db.ts' } + ); + await run(['migrate', 'generate']); + expect(commands.generate).toHaveBeenLastCalledWith( + 'migration', + config, + {} + ); + }); + + it('routes push and tolerates cleanup failures', async () => { + destroy.mockRejectedValue(new Error('pool already closed')); + await run(['db', 'push', '--yes']); + expect(commands.push).toHaveBeenCalledWith(config, { '--yes': true }); + expect(process.exit).not.toHaveBeenCalled(); + }); + + it.each([new Error('database offline'), 'database offline'])( + 'reports command failures after closing the pool', + async error => { + commands.validate.mockRejectedValue(error); + await expect(run(['validate'])).rejects.toThrow('process exited'); + expect(destroy).toHaveBeenCalledOnce(); + expect(console.error).toHaveBeenCalledWith( + expect.stringContaining('database offline') + ); + expect(process.exit).toHaveBeenCalledWith(1); + } + ); + + it('reports configuration failures without a pool', async () => { + commands.load.mockRejectedValue(new Error('configuration missing')); + await expect(run(['validate'])).rejects.toThrow('process exited'); + expect(destroy).not.toHaveBeenCalled(); + }); + + it.each([ + ['unknown'], + ['migrate'], + ['migrate', 'unknown'], + ['db'], + ['db', 'unknown'] + ])('rejects unknown commands: %j', async (...args) => { + await expect(run(args)).rejects.toThrow('process exited'); + expect(console.error).toHaveBeenCalledWith( + expect.stringContaining('Unknown') + ); + expect(commands.load).not.toHaveBeenCalled(); + }); +}); diff --git a/libs/orm-cli/src/commands/push.test.ts b/libs/orm-cli/src/commands/push.test.ts new file mode 100644 index 00000000..580e21c1 --- /dev/null +++ b/libs/orm-cli/src/commands/push.test.ts @@ -0,0 +1,132 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { push } from './push.js'; + +const mocks = vi.hoisted(() => ({ + exists: vi.fn(), + create: vi.fn(), + introspect: vi.fn(), + diff: vi.fn(), + empty: vi.fn(), + apply: vi.fn(), + snapshot: vi.fn(), + write: vi.fn(), + variants: vi.fn(), + prompt: vi.fn(), + close: vi.fn() +})); +vi.mock('@cleverbrush/knex-schema', () => ({ + getTableName: (s: any) => s.table, + getPolymorphicVariantSchemas: mocks.variants, + tableExistsInDb: mocks.exists, + generateCreateTable: () => mocks.create, + introspectDatabase: mocks.introspect, + diffSchema: mocks.diff, + isDiffEmpty: mocks.empty, + applyDiff: mocks.apply, + entitiesToSnapshot: mocks.snapshot, + writeSnapshot: mocks.write +})); +vi.mock('node:readline', () => ({ + default: { + createInterface: () => ({ question: mocks.prompt, close: mocks.close }) + } +})); + +describe('schema push', () => { + const trx = {}; + const transaction = vi.fn(async callback => callback(trx)); + const schema = { table: 'items' }; + const config = { + knex: { transaction }, + entities: { items: { schema } }, + migrations: { directory: './migrations' } + } as any; + beforeEach(() => { + vi.clearAllMocks(); + vi.stubEnv('NODE_ENV', 'test'); + vi.spyOn(console, 'log').mockImplementation(() => {}); + mocks.variants.mockReturnValue([]); + mocks.exists.mockResolvedValue(false); + mocks.empty.mockReturnValue(true); + mocks.snapshot.mockReturnValue({ version: 1, tables: {} }); + }); + afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllEnvs(); + }); + + it('creates new tables and snapshots only after the transaction succeeds', async () => { + await push(config, { '--yes': true }); + expect(mocks.create).toHaveBeenCalledWith(trx); + expect(mocks.write).toHaveBeenCalledWith( + expect.stringContaining('/migrations/snapshot.json'), + { version: 1, tables: {} } + ); + expect(mocks.write.mock.invocationCallOrder[0]).toBeGreaterThan( + mocks.create.mock.invocationCallOrder[0] + ); + expect(mocks.prompt).not.toHaveBeenCalled(); + }); + + it('deduplicates shared tables and applies only non-empty diffs', async () => { + mocks.variants.mockReturnValue([ + { getExtension: () => 'items' }, + { getExtension: () => 'details' }, + { getExtension: () => undefined } + ]); + mocks.exists.mockResolvedValue(true); + mocks.empty.mockReturnValueOnce(false).mockReturnValueOnce(true); + mocks.diff.mockReturnValue({ addColumns: ['name'] }); + await push( + { + ...config, + migrations: { directory: '.', snapshot: './custom.json' } + }, + { '--yes': true } + ); + expect(mocks.exists).toHaveBeenCalledTimes(2); + expect(mocks.apply).toHaveBeenCalledWith( + trx, + { addColumns: ['name'] }, + 'items' + ); + expect(mocks.write).toHaveBeenCalledWith( + expect.stringContaining('/custom.json'), + expect.any(Object) + ); + }); + + it.each(['y', 'YES'])( + 'accepts confirmation %s and reports an unchanged schema', + async answer => { + mocks.prompt.mockImplementation((_question, callback) => + callback(answer) + ); + mocks.exists.mockResolvedValue(true); + await push(config, {}); + expect(mocks.close).toHaveBeenCalledOnce(); + expect(console.log).toHaveBeenCalledWith( + 'No schema changes detected.' + ); + expect(mocks.apply).not.toHaveBeenCalled(); + } + ); + + it('does not access the database or snapshot after rejected confirmation', async () => { + mocks.prompt.mockImplementation((_question, callback) => + callback('no') + ); + await push(config, {}); + expect(transaction).not.toHaveBeenCalled(); + expect(mocks.write).not.toHaveBeenCalled(); + expect(console.log).toHaveBeenCalledWith('Aborted.'); + }); + + it('does not advance the snapshot after a failed write', async () => { + mocks.create.mockRejectedValueOnce(new Error('DDL failed')); + await expect(push(config, { '--yes': true })).rejects.toThrow( + 'DDL failed' + ); + expect(mocks.write).not.toHaveBeenCalled(); + }); +}); diff --git a/libs/orm-cli/src/commands/validate.test.ts b/libs/orm-cli/src/commands/validate.test.ts new file mode 100644 index 00000000..1a9e2545 --- /dev/null +++ b/libs/orm-cli/src/commands/validate.test.ts @@ -0,0 +1,49 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { validate } from './validate.js'; + +const check = vi.hoisted(() => vi.fn()); +vi.mock('@cleverbrush/knex-schema', () => ({ + validateEntitiesAgainstDatabase: check +})); +const keys = [ + 'addColumns', + 'dropColumns', + 'alterColumns', + 'addIndexes', + 'dropIndexes', + 'addForeignKeys', + 'dropForeignKeys' +]; +describe('drift diagnostics', () => { + beforeEach(() => vi.spyOn(console, 'error').mockImplementation(() => {})); + afterEach(() => vi.restoreAllMocks()); + it.each([0, 1, 2])( + 'summarizes all change kinds at count %s', + async count => { + const diff = Object.fromEntries( + keys.map(key => [key, Array(count).fill({})]) + ); + check.mockResolvedValue({ + valid: false, + issues: [{ type: 'schema-drift', tableName: 'items', diff }], + checkedTables: ['items'] + }); + await expect( + validate({ knex: {}, entities: {} } as any) + ).rejects.toThrow('Schema drift detected'); + const output = vi.mocked(console.error).mock.calls.flat().join(' '); + if (!count) expect(output).toContain('unknown drift'); + else { + expect(output).toContain( + `${count} ${count === 1 ? 'column' : 'columns'} to add` + ); + expect(output).toContain( + `${count} ${count === 1 ? 'index' : 'indexes'} to drop` + ); + expect(output).toContain( + `${count} ${count === 1 ? 'foreign key' : 'foreign keys'} to add` + ); + } + } + ); +}); diff --git a/libs/orm/README.md b/libs/orm/README.md index 3be4a538..6d3852f6 100644 --- a/libs/orm/README.md +++ b/libs/orm/README.md @@ -1,6 +1,6 @@ # `@cleverbrush/orm` -![Unit coverage](https://img.shields.io/badge/unit_coverage-81%25-green) +![Unit coverage](https://img.shields.io/badge/unit_coverage-92.1%25-brightgreen) EF-Core-like typed ORM layer on top of [`@cleverbrush/knex-schema`](../knex-schema). diff --git a/libs/orm/src/change-tracker.test.ts b/libs/orm/src/change-tracker.test.ts new file mode 100644 index 00000000..bbc1c609 --- /dev/null +++ b/libs/orm/src/change-tracker.test.ts @@ -0,0 +1,163 @@ +import { date, number, object, string } from '@cleverbrush/knex-schema'; +import { afterEach, describe, expect, it } from 'vitest'; +import { mockDriver } from '../../knex-schema/testing/mock-driver.js'; +import { ChangeTracker } from './change-tracker.js'; +import { ConcurrencyError } from './errors.js'; + +const Row = object({ + id: number().primaryKey(), + name: string().hasColumnName('display_name'), + version: number().rowVersion(), + extra: string().optional() +}).hasTableName('rows'); +const drivers: ReturnType[] = []; +function setup(schema = Row) { + const driver = mockDriver(); + drivers.push(driver); + const tracker = new ChangeTracker(); + tracker.registerEntitySet({ entitySetKey: 'rows', schema }); + driver.respond = q => + q.sql.startsWith('update') || q.sql.startsWith('delete') + ? [{}] + : q.sql.startsWith('select') + ? [{ id: 1, name: 'saved', version: 2, extra: null }] + : q.sql.startsWith('insert') + ? [{ id: 1, display_name: 'saved', version: 2, extra: null }] + : []; + return { driver, tracker }; +} +afterEach(async () => { + await Promise.all(drivers.splice(0).map(d => d.knex.destroy())); +}); + +describe('tracked entity transitions', () => { + it('rejects unknown sets and untracked operations, tolerating untracked detach/reload', async () => { + const { driver, tracker } = setup(); + expect(() => tracker.attach('missing', {})).toThrow( + 'unknown entity set' + ); + expect(() => tracker.entry({})).toThrow('not tracked'); + expect(() => tracker.remove({})).toThrow('not tracked'); + tracker.detach({}); + await tracker.reload({}, driver.knex); + expect(await tracker.saveChanges(driver.knex)).toEqual({ + inserted: 0, + updated: 0, + deleted: 0 + }); + expect(driver.queries).toEqual([]); + }); + + it('deduplicates transient attaches, clears/discards entries, and retains identity', () => { + const { tracker } = setup(); + const added = { name: 'new', version: 1 }; + tracker.attach('rows', added); + tracker.attach('rows', added); + expect(tracker.pendingSummary()).toBe('(1 Added)'); + expect(tracker.entry(added).currentValues).toBe(added); + tracker.entry(added).reset(); + expect(tracker.entry(added).state).toBe('Added'); + tracker.detach(added); + expect(() => tracker.entry(added)).toThrow(); + const row = tracker.attach('rows', { id: 1, name: 'a', version: 1 }); + expect( + tracker.attach('rows', { id: 1, name: 'other', version: 2 }) + ).toBe(row); + row.name = 'changed'; + tracker.attach('rows', row); + expect(tracker.entry(row).originalValues.name).toBe('changed'); + tracker.attach('rows', added); + tracker.discardChanges(); + expect(() => tracker.entry(added)).toThrow(); + tracker.remove(row); + tracker.discardChanges(); + expect(tracker.entry(row).state).toBe('Unchanged'); + tracker.attach('rows', added); + tracker.clear(); + expect(() => tracker.entry(row)).toThrow(); + expect(() => tracker.entry(added)).toThrow(); + }); + + it('commits generated IDs before refreshing snapshots and can delete the new entity', async () => { + const { driver, tracker } = setup(); + const row = tracker.attach('rows', { name: 'new', version: 1 } as any); + expect(await tracker.saveChanges(driver.knex)).toEqual({ + inserted: 1, + updated: 0, + deleted: 0 + }); + expect(row).toMatchObject({ id: 1, name: 'saved', version: 2 }); + expect(tracker.attach('rows', { id: 1 })).toBe(row); + tracker.remove(row); + expect(await tracker.saveChanges(driver.knex)).toEqual({ + inserted: 0, + updated: 0, + deleted: 1 + }); + expect( + driver.queries.find(q => q.sql.startsWith('delete'))?.bindings + ).toEqual([1, 2]); + }); + + it('does not commit version or snapshot changes after a failed transaction', async () => { + const { driver, tracker } = setup(); + const row = tracker.attach('rows', { + id: 1, + name: 'old', + version: 1, + extra: undefined + }); + row.name = 'new'; + row.extra = 'added'; + driver.respond = () => []; + await expect(tracker.saveChanges(driver.knex)).rejects.toBeInstanceOf( + ConcurrencyError + ); + expect(row.version).toBe(1); + expect(tracker.entry(row).originalValues.name).toBe('old'); + expect(driver.queries.at(-1)?.sql).toContain('ROLLBACK'); + driver.respond = q => (q.sql.startsWith('update') ? [{}] : []); + await tracker.saveChanges(driver.knex); + expect(row.version).toBe(2); + expect(tracker.entry(row).isModified()).toBe(false); + }); + + it('reloads mapped columns/version and tolerates missing rows', async () => { + const { driver, tracker } = setup(); + const row = tracker.attach('rows', { id: 1, name: 'old', version: 1 }); + row.name = 'unsaved'; + await tracker.reload(row, driver.knex); + expect(row).toMatchObject({ name: 'saved', version: 2 }); + driver.respond = () => []; + await tracker.reload(row, driver.knex); + expect(row.name).toBe('saved'); + }); + + it('guards deletes with row versions and rolls back concurrency failures', async () => { + const { driver, tracker } = setup(); + const row = tracker.attach('rows', { id: 1, name: 'old', version: 1 }); + tracker.remove(row); + driver.respond = () => []; + await expect(tracker.saveChanges(driver.knex)).rejects.toBeInstanceOf( + ConcurrencyError + ); + expect(tracker.entry(row).state).toBe('Deleted'); + }); + + it('generates timestamp versions only on successful updates', async () => { + const schema = object({ + id: number().primaryKey(), + name: string(), + version: date().rowVersion() + }).hasTableName('rows'); + const { driver, tracker } = setup(schema as any); + const version = new Date('2020-01-01'); + const row = tracker.attach('rows', { id: 1, name: 'old', version }); + row.name = 'new'; + await tracker.saveChanges(driver.knex); + expect(row.version.getTime()).toBeGreaterThan(version.getTime()); + expect( + driver.queries.find(q => q.sql.startsWith('update'))?.bindings + ).toContain(version); + }); +}); diff --git a/libs/orm/src/dbset.tracking.test.ts b/libs/orm/src/dbset.tracking.test.ts new file mode 100644 index 00000000..d459d8b9 --- /dev/null +++ b/libs/orm/src/dbset.tracking.test.ts @@ -0,0 +1,205 @@ +import { + date, + defineEntity, + number, + object, + parameter, + string +} from '@cleverbrush/knex-schema'; +import { afterEach, describe, expect, it } from 'vitest'; +import { mockDriver } from '../../knex-schema/testing/mock-driver.js'; +import { ChangeTracker } from './change-tracker.js'; +import { createDb } from './dbcontext.js'; + +const Plain = defineEntity( + object({ id: number().primaryKey(), name: string() }).hasTableName('items') +); +const Poly = defineEntity( + object({ + id: number().primaryKey(), + kind: string(), + version: number().rowVersion() + }).hasTableName('assets') +) + .discriminator('kind') + .stiVariant('note', object({ text: string() })); +const drivers: ReturnType[] = []; +function setup() { + const driver = mockDriver(); + drivers.push(driver); + driver.respond = q => + q.sql.includes('count(') + ? [{ count: '1' }] + : q.sql.includes('cast("id" as text)') + ? [{ id: '1' }] + : q.sql.includes('__read_poly') + ? [ + { + __read_poly: { + id: 1, + kind: 'note', + text: 'stored', + version: 1 + } + } + ] + : /^(select|with)/.test(q.sql) + ? [{ id: 1, name: 'stored', __cursor_value: 1 }] + : /^(insert|update)/.test(q.sql) + ? [{ id: 1, kind: 'note', text: 'stored', version: 1 }] + : q.sql.startsWith('delete') + ? [{}] + : []; + return { + driver, + db: createDb( + driver.knex, + { items: Plain, assets: Poly }, + { tracking: true } + ) + }; +} +afterEach(async () => { + await Promise.all(drivers.splice(0).map(d => d.knex.destroy())); +}); +describe('tracked query result boundaries', () => { + it('preserves identities across awaited, callable, first and paginated reads', async () => { + const { db } = setup(); + const [row] = await db.items; + const read = db.items.where('id', parameter('id')); + expect((await read(1))[0]).toBe(row); + expect(await db.items.first()).toBe(row); + expect((await db.items.execute())[0]).toBe(row); + expect( + (await db.items.paginate({ page: 1, pageSize: 1 })).data[0] + ).toBe(row); + expect((await db.items.paginateAfter({ limit: 1 })).data[0]).toBe(row); + expect(await db.items.pluck('id')).toEqual([1]); + for (const operation of [ + 'find', + 'findMany', + 'insert', + 'update', + 'delete', + 'restore', + 'hardDelete' + ]) + expect(() => (read as any)[operation]({})).toThrow( + /parameter|bind|argument/i + ); + }); + it('preserves variant identities across callable, scalar and paginated queries', async () => { + const { db } = setup(); + const view = db.assets.ofVariant('note'); + const [row] = await view; + expect((await view.execute())[0]).toBe(row); + expect(await view.first()).toBe(row); + expect((await view.where('id', parameter('id'))(1))[0]).toBe(row); + expect((await view.paginate({ page: 1, pageSize: 1 })).data[0]).toBe( + row + ); + expect(await view.pluck('id')).toEqual([1]); + expect(view.rowSchema).toBeDefined(); + const projected = view.forVariant('note', q => + q.select(t => ({ id: t.id, kind: t.kind })) + ); + expect((await projected.execute())[0]).not.toBe(row); + expect(() => db.entry({})).toThrow('not tracked'); + const unbound = view.where('id', parameter('id')); + expect(() => (unbound as any).find(1)).toThrow( + /parameter|bind|argument/i + ); + }); + it('keeps transaction-bound variant views and their tracking wrappers', async () => { + const { db } = setup(); + await db.transaction(async transactional => { + const view = transactional.assets.ofVariant('note'); + expect(await view.find(1)).toMatchObject({ kind: 'note' }); + expect(await view.findMany([1])).toHaveLength(1); + }); + const { driver } = setup(); + await driver.knex.transaction(async trx => { + const view = db.assets.ofVariant('note').withTransaction(trx); + expect((await view.execute())[0].kind).toBe('note'); + }); + }); + it('persists polymorphic tracked updates/deletes and guards optimistic versions', async () => { + const { driver } = setup(); + const tracker = new ChangeTracker(); + tracker.registerEntitySet({ + entitySetKey: 'assets', + schema: Poly.schema + }); + const row = tracker.attach('assets', { + id: 1, + kind: 'note', + text: 'old', + version: 1 + }); + row.text = 'new'; + expect(await tracker.saveChanges(driver.knex)).toMatchObject({ + updated: 1 + }); + expect(driver.queries.some(q => q.sql.startsWith('update'))).toBe(true); + tracker.remove(row); + expect(await tracker.saveChanges(driver.knex)).toMatchObject({ + deleted: 1 + }); + const other = tracker.attach('assets', { + id: 2, + kind: 'note', + text: 'other', + version: 1 + }); + other.text = 'changed'; + driver.respond = () => []; + await expect(tracker.saveChanges(driver.knex)).rejects.toThrow( + /concurr|version/i + ); + tracker.discardChanges(); + tracker.remove(other); + await expect(tracker.saveChanges(driver.knex)).rejects.toThrow( + /concurr|version/i + ); + }); + it('tracks deep document mutations, resets independent copies and validates snapshots', async () => { + const { driver } = setup(); + const document = object({ + tags: object({ name: string() }), + created: date().optional() + }) + .acceptUnknownProps() + .jsonb(); + const schema = object({ + id: number().primaryKey(), + data: document + }).hasTableName('documents'); + const tracker = new ChangeTracker(); + tracker.registerEntitySet({ entitySetKey: 'documents', schema }); + const row = tracker.attach('documents', { + id: 1, + data: { + tags: { name: 'a' }, + created: new Date(0), + extra: [1, null] + } + }); + row.data.tags.name = 'b'; + expect(tracker.entry(row).isModified('data')).toBe(true); + tracker.entry(row).reset(); + expect(row.data.tags.name).toBe('a'); + row.data.extra.push(2); + expect(tracker.hasPendingChanges()).toBe(true); + await tracker.saveChanges(driver.knex); + expect(tracker.entry(row).isModified()).toBe(false); + row.data.extra = [1, null]; + tracker.discardChanges(); + expect(row.data.extra).toEqual([1, null, 2]); + expect(() => + tracker.attach('documents', { + id: 3, + data: { tags: { name: 'a' }, extra: NaN } + }) + ).toThrow(); + }); +}); diff --git a/libs/react-form/README.md b/libs/react-form/README.md index 8cb45134..f0443fea 100644 --- a/libs/react-form/README.md +++ b/libs/react-form/README.md @@ -3,7 +3,7 @@ [![CI](https://github.com/cleverbrush/framework/actions/workflows/ci.yml/badge.svg)](https://github.com/cleverbrush/framework/actions/workflows/ci.yml) [![License: BSD-3-Clause](https://img.shields.io/badge/license-BSD--3--Clause-blue.svg)](../../LICENSE) -![Unit coverage](https://img.shields.io/badge/unit_coverage-95.5%25-brightgreen) +![Unit coverage](https://img.shields.io/badge/unit_coverage-99.5%25-brightgreen) A headless, schema-driven form system for React based on `@cleverbrush/schema`. Uses PropertyDescriptors for type-safe field binding, supports global UI renderer configuration via a provider, and is completely UI-agnostic — works with plain HTML, MUI, Ant Design, or any component library. diff --git a/libs/react-form/src/FormStore.test.ts b/libs/react-form/src/FormStore.test.ts index 013a83d4..7d6b3bee 100644 --- a/libs/react-form/src/FormStore.test.ts +++ b/libs/react-form/src/FormStore.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, test } from 'vitest'; +import { describe, expect, test, vi } from 'vitest'; import { createFormStore } from './FormStore.js'; function setup(profile: any) { @@ -72,3 +72,28 @@ describe('shared deep utilities in the form store', () => { expect(store.getFieldState('/profile').dirty).toBe(true); }); }); +test('notifies subscribers explicitly and clears root issues after value changes', () => { + const store = createFormStore(null); + const listener = vi.fn(); + const unsubscribe = store.subscribe('/name', listener); + store.setIssues([{ pointer: '', detail: 'Invalid form' }]); + expect(store.getFieldState('/name').error).toBeUndefined(); + store.notifyAll(); + expect(listener).toHaveBeenCalled(); + store.setValues({ name: 'Ada' }); + expect(store.getSubmissionState().error).toBeUndefined(); + unsubscribe(); +}); + +test('safely reads escaped and malformed pointers when invalidating server issues', () => { + const store = createFormStore({ 'a/b': { 'c~d': 'old' }, name: 'old' }); + store.setIssues([ + { pointer: '/a~1b/c~0d', detail: 'Reserved value' }, + { pointer: '/missing/child', detail: 'Unknown field' }, + { pointer: '/bad~2pointer', detail: 'Malformed escape' }, + { pointer: 'not-a-pointer', detail: 'Malformed root' } + ]); + store.setValues({ 'a/b': { 'c~d': 'new' }, name: 'new' }); + expect(store.getSubmissionState().error).not.toContain('Reserved value'); + expect(store.getSubmissionState().error).toContain('Malformed escape'); +}); diff --git a/libs/react-form/src/debounce.test.ts b/libs/react-form/src/debounce.test.ts new file mode 100644 index 00000000..812aef47 --- /dev/null +++ b/libs/react-form/src/debounce.test.ts @@ -0,0 +1,25 @@ +import { afterEach, expect, it, vi } from 'vitest'; +import { debounce } from './debounce.js'; + +afterEach(() => vi.useRealTimers()); + +it('delays execution and replaces both the pending arguments and deadline', () => { + vi.useFakeTimers(); + const callback = vi.fn(); + const debounced = debounce(callback, 100); + debounced('first', 1); + vi.advanceTimersByTime(75); + expect(callback).not.toHaveBeenCalled(); + debounced('second', 2); + vi.advanceTimersByTime(99); + expect(callback).not.toHaveBeenCalled(); + vi.advanceTimersByTime(1); + expect(callback).toHaveBeenCalledExactlyOnceWith('second', 2); + debounced('third', 3); + vi.advanceTimersByTime(100); + expect(callback.mock.calls).toEqual([ + ['second', 2], + ['third', 3] + ]); + expect(vi.getTimerCount()).toBe(0); +}); diff --git a/libs/react-form/src/helpers.test.ts b/libs/react-form/src/helpers.test.ts new file mode 100644 index 00000000..d1814f6e --- /dev/null +++ b/libs/react-form/src/helpers.test.ts @@ -0,0 +1,40 @@ +import { object, string } from '@cleverbrush/schema'; +import { expect, it } from 'vitest'; +import { + buildDescriptorPathMap, + buildSelectorFromPath, + ensureNestedStructure, + getDescriptorPath, + getSchemaType, + isErrorPathMatch +} from './helpers.js'; + +it('handles absent descriptor paths and incomplete external descriptor trees', () => { + const schema = object({ + name: string(), + nested: object({ city: string() }) + }); + expect(buildDescriptorPathMap({} as any, schema).size).toBe(0); + expect(getDescriptorPath({} as any, new Map())).toBe(''); + expect(buildSelectorFromPath('nested.city')({})).toBeUndefined(); + expect( + buildSelectorFromPath('nested.city')({ nested: { city: 'Paris' } }) + ).toBe('Paris'); + expect(getSchemaType({ introspect: () => ({}) } as any)).toBe('unknown'); + expect(ensureNestedStructure(null, schema)).toEqual({ nested: {} }); +}); + +it('matches only the field itself, its validators and nested fields', () => { + for (const path of ['$.name', '$.name.child', '$.name($validators[0])']) + expect(isErrorPathMatch(path, '$.name')).toBe(true); + expect(isErrorPathMatch('$.names', '$.name')).toBe(false); +}); + +it('handles introspection adapters without object properties', () => { + const external = { introspect: () => ({ type: 'object' }) } as any; + expect(buildDescriptorPathMap({} as any, external).size).toBe(0); + expect(ensureNestedStructure(null, external)).toEqual({}); + expect(ensureNestedStructure(undefined, external)).toEqual({}); + const values = { name: 'Ada' }; + expect(ensureNestedStructure(values, external)).toBe(values); +}); diff --git a/libs/react-form/src/validation-failures.test.tsx b/libs/react-form/src/validation-failures.test.tsx new file mode 100644 index 00000000..1cbce22c --- /dev/null +++ b/libs/react-form/src/validation-failures.test.tsx @@ -0,0 +1,47 @@ +import { object, string } from '@cleverbrush/schema'; +import { act, cleanup, renderHook } from '@testing-library/react'; +import { afterEach, expect, it, vi } from 'vitest'; +import { useSchemaForm } from './hooks.js'; + +afterEach(() => { + cleanup(); + vi.restoreAllMocks(); +}); +it('settles field validation when a schema rejects unexpectedly', async () => { + const schema = object({ name: string() }); + vi.spyOn(schema, 'validateAsync').mockRejectedValue( + new Error('validator unavailable') + ); + const hook = renderHook(() => { + const form = useSchemaForm(schema); + return { form, field: form.useField(t => t.name) }; + }); + await act(() => hook.result.current.form.validate()); + expect(hook.result.current.field.validating).toBe(false); + const save = vi.fn(); + await act(() => hook.result.current.form.handleSubmit(save)()); + expect(save).not.toHaveBeenCalled(); + expect(hook.result.current.form.submitting).toBe(false); +}); +it('ignores a rejected submission after unmount without calling its error handler', async () => { + const hook = renderHook(() => useSchemaForm(object({ name: string() }))); + act(() => hook.result.current.reset({ name: 'Ada' })); + let reject!: (error: Error) => void; + const save = vi.fn( + () => + new Promise((_resolve, fail) => { + reject = fail; + }) + ); + const onError = vi.fn(); + let submit!: Promise; + await act(async () => { + submit = hook.result.current.handleSubmit(save, { onError })(); + await Promise.resolve(); + }); + expect(save).toHaveBeenCalledOnce(); + hook.unmount(); + reject(new Error('late failure')); + await submit; + expect(onError).not.toHaveBeenCalled(); +}); diff --git a/libs/scheduler-postgres/README.md b/libs/scheduler-postgres/README.md index 9ef51c5d..1baee3f1 100644 --- a/libs/scheduler-postgres/README.md +++ b/libs/scheduler-postgres/README.md @@ -1,6 +1,6 @@ # @cleverbrush/scheduler-postgres -![Unit coverage](https://img.shields.io/badge/unit_coverage-11.2%25-red) +![Unit coverage](https://img.shields.io/badge/unit_coverage-100%25-brightgreen) PostgreSQL persistence for [@cleverbrush/scheduler](../scheduler/README.md). diff --git a/libs/scheduler-postgres/src/migrations.test.ts b/libs/scheduler-postgres/src/migrations.test.ts new file mode 100644 index 00000000..161cebcb --- /dev/null +++ b/libs/scheduler-postgres/src/migrations.test.ts @@ -0,0 +1,50 @@ +import { expect, it } from 'vitest'; +import { mockDriver } from '../../knex-schema/testing/mock-driver.js'; +import { createSchedulerTables, dropSchedulerTables } from './migrations.js'; + +it('creates namespaced tables, indexes and cascade constraints, then drops in dependency order', async () => { + const { knex, queries } = mockDriver(); + try { + await createSchedulerTables(knex, { tablePrefix: 'unit_jobs' }); + const sql = queries.map(q => q.sql).join('\n'); + for (const name of ['runs', 'schedules', 'events', 'attempts']) + expect(sql).toContain(`create table "unit_jobs_${name}"`); + expect(sql).toContain('on delete CASCADE'); + expect(sql).toContain('"unit_jobs_runs_dedupe"'); + expect(sql).toContain('"unit_jobs_runs_ready"'); + queries.length = 0; + await dropSchedulerTables(knex, { tablePrefix: 'unit_jobs' }); + expect( + queries.filter(q => q.sql.startsWith('drop table')).map(q => q.sql) + ).toEqual([ + 'drop table "unit_jobs_events"', + 'drop table "unit_jobs_attempts"', + 'drop table "unit_jobs_schedules"', + 'drop table "unit_jobs_runs"' + ]); + } finally { + await knex.destroy(); + } +}); + +it('rolls back a failed migration and validates prefixes before opening a transaction', async () => { + const { knex, queries } = mockDriver(); + const failing = mockDriver(); + failing.respond = query => { + if (query.sql.startsWith('create table')) throw new Error('DDL failed'); + return []; + }; + try { + await expect(createSchedulerTables(failing.knex)).rejects.toThrow( + 'DDL failed' + ); + expect(failing.queries.at(-1)?.sql).toBe('ROLLBACK'); + await expect( + createSchedulerTables(knex, { tablePrefix: 'invalid-prefix' }) + ).rejects.toThrow('Invalid scheduler table prefix'); + expect(queries).toEqual([]); + } finally { + await knex.destroy(); + await failing.knex.destroy(); + } +}); diff --git a/libs/scheduler-postgres/src/storage.test.ts b/libs/scheduler-postgres/src/storage.test.ts new file mode 100644 index 00000000..33fa3d95 --- /dev/null +++ b/libs/scheduler-postgres/src/storage.test.ts @@ -0,0 +1,258 @@ +import { + InMemoryJobRepository, + JobScheduler, + type RunRecord, + type ScheduleRecord +} from '@cleverbrush/scheduler'; +import { afterEach, describe, expect, it } from 'vitest'; +import { mockDriver } from '../../knex-schema/testing/mock-driver.js'; +import { testJob } from '../../scheduler/testing/repository-contract.js'; +import { PostgresJobRepository, PostgresJobStorage } from './storage.js'; + +const drivers: ReturnType[] = []; +afterEach(async () => { + await Promise.all(drivers.splice(0).map(d => d.knex.destroy())); +}); +function setup() { + const driver = mockDriver(); + drivers.push(driver); + return { + ...driver, + driver, + storage: new PostgresJobStorage(driver.knex, { + tablePrefix: 'unit_jobs' + }) + }; +} +async function runRecord(): Promise { + const repo = new InMemoryJobRepository(); + const scheduler = new JobScheduler({ storageRepository: repo }); + const run = await scheduler.enqueue(testJob(), { id: 'one' }); + return (await repo.get('default', run.id))!; +} +const row = (run: RunRecord) => ({ + id: run.id, + namespace: run.namespace, + name: run.name, + version: run.version, + status: run.status, + availableAt: run.availableAt, + leaseExpiresAt: run.leaseExpiresAt, + scheduleId: run.scheduleId, + dedupeKey: run.dedupeKey, + expiresAt: null, + record: JSON.stringify(run) +}); + +describe('PostgreSQL storage SQL and transaction contract', () => { + it('sets transaction timeouts, uses database time and rolls back failures', async () => { + const { driver, storage, queries } = setup(); + driver.respond = query => + query.sql.includes(' as now') ? [{ now: 1234 }] : []; + expect(await storage.atomic(tx => tx.now())).toBe(1234); + expect(queries.map(q => q.sql)).toEqual( + expect.arrayContaining([ + 'BEGIN;', + "set local lock_timeout = '5s'", + "set local statement_timeout = '15s'", + 'COMMIT;' + ]) + ); + await expect( + storage.atomic(async () => { + throw new Error('rollback'); + }) + ).rejects.toThrow('rollback'); + expect(queries.at(-1)?.sql).toBe('ROLLBACK'); + }); + + it('reads scoped records, returns missing rows and locks requested reads', async () => { + const { driver, storage, queries } = setup(); + const record = await runRecord(); + driver.respond = query => + query.sql.startsWith('select') ? [row(record)] : []; + for (const lock of [false, true]) { + expect( + await storage.atomic(tx => tx.run('tenant', record.id, lock)) + ).toEqual(record); + const read = queries.findLast(q => q.sql.startsWith('select'))!; + expect(read.bindings).toEqual( + expect.arrayContaining(['tenant', record.id]) + ); + expect(read.sql.includes('for update')).toBe(lock); + } + driver.respond = () => []; + expect( + await storage.atomic(tx => tx.run('tenant', 'missing', false)) + ).toBeUndefined(); + }); + + it('handles inserts, deduplication races, retention and missing conflicts', async () => { + const { driver, storage, queries } = setup(); + const record = await runRecord(); + driver.respond = query => + query.sql.startsWith('insert') ? [row(record)] : []; + expect(await storage.atomic(tx => tx.insertRun(record))).toEqual({ + record, + inserted: true + }); + const insert = queries.find(q => q.sql.startsWith('insert'))!; + expect(insert.sql).toContain( + 'on conflict ("namespace", "dedupeKey") do nothing' + ); + expect(insert.bindings).toContain(JSON.stringify(record)); + driver.respond = query => + query.sql.startsWith('select') ? [row(record)] : []; + expect(await storage.atomic(tx => tx.insertRun(record))).toEqual({ + record, + inserted: false + }); + await storage.atomic(tx => tx.saveRun({ ...record, completedAt: 100 })); + expect( + queries.findLast(q => q.sql.startsWith('update'))?.bindings + ).toContain(100 + record.policy.retentionMs); + driver.respond = () => []; + await expect( + storage.atomic(tx => tx.insertRun(record)) + ).rejects.toThrow('Conflicting run disappeared'); + }); + + it('claims expired leases and supported versions without blocking locked rows', async () => { + const { driver, storage, queries } = setup(); + const record = await runRecord(); + driver.respond = query => + query.sql.startsWith('select') ? [row(record)] : []; + for (const supported of [[], [{ name: 'report', version: 1 }]]) { + expect( + await storage.atomic(tx => tx.runnable('tenant', supported)) + ).toEqual(record); + const query = queries.findLast(q => q.sql.startsWith('select'))!; + expect(query.sql).toContain('for update skip locked'); + expect(query.sql).toContain('clock_timestamp()'); + expect(query.bindings).toContain('tenant'); + expect(query.sql.includes('"name" =')).toBe(supported.length > 0); + } + }); + + it('serializes progress and attempts and reads them in cursor order', async () => { + const { driver, storage, queries } = setup(); + const event = { + runId: 'r', + sequence: 2, + type: 'progress', + data: { percent: 50 } + } as any; + const attempt = { runId: 'r', attempt: 1, status: 'succeeded' } as any; + driver.respond = query => + query.sql.startsWith('select') + ? [ + { + runId: 'r', + sequence: 2, + attempt: 1, + record: JSON.stringify( + query.sql.includes('_events') ? event : attempt + ) + } + ] + : []; + await storage.atomic(async tx => { + await tx.appendEvent(event); + await tx.saveAttempt(attempt); + expect(await tx.events('r', 1, 10)).toEqual([event]); + expect(await tx.attempts('r')).toEqual([attempt]); + }); + const events = queries.find( + q => q.sql.startsWith('select') && q.sql.includes('_events') + )!; + expect(events.sql).toContain('"sequence" >'); + expect(events.sql).toMatch(/order by .*"sequence" asc limit/); + expect(events.bindings).toEqual(['r', 1, 10]); + expect( + queries.find( + q => q.sql.startsWith('insert') && q.sql.includes('_attempts') + )?.sql + ).toContain('do update'); + }); + + it('locks schedule cursors, limits due work and detects unfinished runs', async () => { + const { driver, storage, queries } = setup(); + const schedule = { + namespace: 'tenant', + id: 'daily', + active: true, + nextAt: 100 + } as ScheduleRecord; + driver.respond = query => + query.sql.startsWith('select') + ? [{ ...schedule, record: JSON.stringify(schedule) }] + : []; + await storage.atomic(async tx => { + expect(await tx.schedule('tenant', 'daily')).toEqual(schedule); + expect(await tx.insertSchedule(schedule)).toEqual(schedule); + await tx.saveSchedule(schedule); + expect(await tx.dueSchedules('tenant', 5)).toEqual([schedule]); + }); + const due = queries.findLast(q => q.sql.startsWith('select'))!; + expect(due.sql).toContain('for update skip locked'); + expect(due.bindings).toEqual(['tenant', true, 5]); + const record = await runRecord(); + driver.respond = query => + query.sql.startsWith('select') ? [row(record)] : []; + expect( + await storage.atomic(tx => + tx.unfinishedScheduleRun('tenant', 'daily') + ) + ).toBe(true); + driver.respond = () => []; + expect( + await storage.atomic(tx => + tx.unfinishedScheduleRun('tenant', 'daily') + ) + ).toBe(false); + }); + + it('bounds cleanup to expired terminal rows and does nothing for empty pages', async () => { + const { driver, storage, queries } = setup(); + expect(await storage.atomic(tx => tx.cleanup('tenant', 2))).toBe(0); + driver.respond = query => + query.sql.startsWith('select') + ? [{ id: 'r' }] + : query.sql.startsWith('delete') + ? [{}] + : []; + expect(await storage.atomic(tx => tx.cleanup('tenant', 2))).toBe(1); + const select = queries.findLast(q => q.sql.startsWith('select'))!; + expect(select.sql).toContain('for update skip locked'); + expect(select.bindings).toEqual([ + 'tenant', + 'succeeded', + 'failed', + 'cancelled', + 2 + ]); + expect( + queries.findLast(q => q.sql.startsWith('delete'))?.bindings + ).toEqual(['r']); + }); + + it('reports status counts, queued definitions and nullable oldest-ready time', async () => { + const { driver, storage } = setup(); + driver.respond = query => { + if (query.sql.includes('min(')) return [{ at: null }]; + if (query.sql.includes('group by "name"')) + return [{ name: 'report', version: 1, count: 2 }]; + if (query.sql.includes('group by "status"')) + return [{ status: 'queued', count: 2 }]; + return []; + }; + expect(await storage.atomic(tx => tx.health('tenant'))).toEqual({ + counts: { queued: 2 }, + oldestReadyAt: null, + queuedDefinitions: [{ name: 'report', version: 1, count: 2 }] + }); + expect(new PostgresJobRepository(driver.knex)).toBeInstanceOf( + PostgresJobRepository + ); + }); +}); diff --git a/libs/scheduler/README.md b/libs/scheduler/README.md index 0ce773b4..11ebc515 100644 --- a/libs/scheduler/README.md +++ b/libs/scheduler/README.md @@ -1,6 +1,6 @@ # @cleverbrush/scheduler -![Unit coverage](https://img.shields.io/badge/unit_coverage-83.4%25-green) +![Unit coverage](https://img.shields.io/badge/unit_coverage-96.3%25-brightgreen) Typed immediate, delayed and recurring jobs with durable progress. Producers, diff --git a/libs/scheduler/src/lifecycle.test.ts b/libs/scheduler/src/lifecycle.test.ts new file mode 100644 index 00000000..fc3a00b8 --- /dev/null +++ b/libs/scheduler/src/lifecycle.test.ts @@ -0,0 +1,231 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { testJob } from '../testing/repository-contract.js'; +import { + InMemoryJobRepository, + JobScheduler, + LeaseLostError +} from './index.js'; +import type { JobWorker } from './worker.js'; + +const workers: JobWorker[] = []; +const schedulers: JobScheduler[] = []; +afterEach(async () => { + await Promise.all( + workers.splice(0).map(w => w.stop({ drainTimeoutMs: 10 })) + ); + await Promise.all(schedulers.splice(0).map(s => s.stop())); + vi.restoreAllMocks(); +}); +function setup() { + const repository = new InMemoryJobRepository(); + const scheduler = new JobScheduler({ + storageRepository: repository, + pollIntervalMs: 2 + }); + schedulers.push(scheduler); + return { repository, scheduler }; +} +describe('durable lifecycle failures', () => { + it('validates worker registrations and single-use lifecycle', async () => { + const { scheduler } = setup(); + const binding = testJob().handle(() => ({ url: '/' })); + expect(() => + scheduler.createWorker({ jobs: [], heartbeatMs: 10, leaseMs: 10 }) + ).toThrow('shorter'); + expect(() => scheduler.createWorker({ jobs: [] })).toThrow( + 'At least one' + ); + expect(() => + scheduler.createWorker({ jobs: [binding, binding] }) + ).toThrow('Duplicate'); + expect(() => + scheduler.createWorker({ + jobs: [{ definition: binding.definition } as any] + }) + ).toThrow('exactly one'); + const worker = scheduler.createWorker({ jobs: [binding] }); + workers.push(worker); + await worker.start(); + await expect(worker.start()).rejects.toThrow('already'); + await worker.stop(); + await expect(worker.start()).rejects.toThrow('already'); + }); + it('isolates dispatch observers and recovers from storage errors', async () => { + const repository = new InMemoryJobRepository(); + const error = new Error('temporarily offline'); + const dispatch = vi + .spyOn(repository, 'dispatch') + .mockRejectedValueOnce(error); + const observer = vi.fn(() => { + throw new Error('observer failed'); + }); + const scheduler = new JobScheduler({ + storageRepository: repository, + pollIntervalMs: 2, + onError: observer + }); + schedulers.push(scheduler); + await scheduler.start(); + await expect(scheduler.start()).rejects.toThrow('already'); + await vi.waitFor(() => expect(observer).toHaveBeenCalledWith(error)); + await vi.waitFor(() => + expect(dispatch.mock.calls.length).toBeGreaterThan(1) + ); + await vi.waitFor(() => expect(scheduler.lastError).toBeUndefined()); + }); + it('validates schedule policies, dates and event cursors and rejects mismatched definitions', async () => { + const { scheduler } = setup(); + const job = testJob(); + await expect( + scheduler.enqueue(job, { id: 'a' }, { runAt: new Date(NaN) }) + ).rejects.toThrow('Invalid runAt'); + await expect( + scheduler.upsertSchedule( + 'daily', + job, + { id: 'a' }, + { + schedule: { type: 'interval', everyMs: 1000 } as any, + missed: 'invalid' as any + } + ) + ).rejects.toThrow('Invalid schedule policy'); + await expect( + scheduler.events(job, 'missing', { after: -1 }).next() + ).rejects.toThrow('Invalid event cursor'); + expect(await scheduler.events(job, 'missing').next()).toMatchObject({ + done: true + }); + const run = await scheduler.enqueue(job, { id: 'a' }); + await expect( + scheduler.getRun({ ...job, version: job.version + 1 }, run.id) + ).rejects.toThrow('does not match'); + const controller = new AbortController(); + const events = scheduler.events(job, run.id, { + signal: controller.signal + }); + expect((await events.next()).done).toBe(false); + controller.abort(); + expect((await events.next()).done).toBe(true); + expect(await scheduler.health()).toHaveProperty('counts'); + expect(await scheduler.cleanup()).toBe(0); + expect(await scheduler.pauseSchedule('missing')).toBe(false); + expect(await scheduler.removeSchedule('missing')).toBe(false); + }); + it.each([new Error('heartbeat unavailable'), new LeaseLostError()])( + 'aborts work after lost heartbeat without committing a terminal state', + async error => { + const { repository, scheduler } = setup(); + const job = testJob({ timeoutMs: 1000 }); + const run = await scheduler.enqueue(job, { id: 'a' }); + const heartbeat = vi + .spyOn(repository, 'heartbeat') + .mockRejectedValue(error); + const complete = vi.spyOn(repository, 'complete'); + const fail = vi.spyOn(repository, 'fail'); + const diagnostic = vi.fn(); + const worker = scheduler.createWorker({ + heartbeatMs: 2, + leaseMs: 1000, + pollIntervalMs: 2, + onDiagnostic: diagnostic, + jobs: [ + job.handle(async (_, context) => { + await new Promise(resolve => + context.signal.addEventListener( + 'abort', + () => resolve(), + { once: true } + ) + ); + return { url: '/' }; + }) + ] + }); + workers.push(worker); + await worker.start(); + await vi.waitFor(() => expect(heartbeat).toHaveBeenCalled()); + await worker.stop({ drainTimeoutMs: 20 }); + expect(complete).not.toHaveBeenCalled(); + expect(fail).not.toHaveBeenCalled(); + expect(diagnostic).toHaveBeenCalledWith( + expect.objectContaining({ + type: + error instanceof LeaseLostError + ? 'lease_lost' + : 'infrastructure_error', + runId: run.id + }) + ); + } + ); + it('returns a claim interrupted by shutdown and isolates failing diagnostics', async () => { + const { repository, scheduler } = setup(); + const job = testJob(); + const run = await scheduler.enqueue(job, { id: 'a' }); + const original = repository.claim.bind(repository); + let release!: () => void; + const claimed = new Promise(resolve => { + release = resolve; + }); + const claim = vi + .spyOn(repository, 'claim') + .mockImplementation(async (...args) => { + const result = await original(...args); + await claimed; + return result; + }); + const handler = vi.fn(() => ({ url: '/' })); + const worker = scheduler.createWorker({ + pollIntervalMs: 2, + jobs: [job.handle(handler)], + onDiagnostic: () => { + throw new Error('ignored'); + } + }); + workers.push(worker); + await worker.start(); + await vi.waitFor(() => expect(claim).toHaveBeenCalled()); + const stopped = worker.stop(); + release(); + await stopped; + expect(handler).not.toHaveBeenCalled(); + expect(await scheduler.getRun(job, run.id)).toMatchObject({ + status: 'failed', + error: { code: 'shutdown' } + }); + }); + it.each([new Error('write unavailable'), new LeaseLostError()])( + 'reports failure-persistence errors without crashing the poll loop', + async error => { + const { repository, scheduler } = setup(); + const job = testJob(); + await scheduler.enqueue(job, { id: 'a' }); + vi.spyOn(repository, 'fail').mockRejectedValue(error); + const diagnostic = vi.fn(() => { + throw new Error('observer unavailable'); + }); + const worker = scheduler.createWorker({ + pollIntervalMs: 2, + jobs: [ + job.handle(() => { + throw new Error('job failed'); + }) + ], + onDiagnostic: diagnostic + }); + workers.push(worker); + await worker.start(); + await vi.waitFor(() => + expect(diagnostic).toHaveBeenCalledWith( + expect.objectContaining({ + type: + error instanceof LeaseLostError + ? 'lease_lost' + : 'infrastructure_error' + }) + ) + ); + } + ); +}); diff --git a/libs/scheduler/src/thread-entry.test.ts b/libs/scheduler/src/thread-entry.test.ts new file mode 100644 index 00000000..4dda5a75 --- /dev/null +++ b/libs/scheduler/src/thread-entry.test.ts @@ -0,0 +1,144 @@ +import { beforeEach, expect, it, vi } from 'vitest'; + +const state = vi.hoisted(() => ({ + handler: vi.fn() as any, + listener: undefined as ((message: any) => void) | undefined, + ackError: undefined as any, + port: { on: vi.fn(), postMessage: vi.fn(), close: vi.fn() }, + data: { + runId: 'run', + attempt: 2, + input: { id: 'one' }, + moduleUrl: '', + policy: { + maxProgressEvents: 2, + maxProgressBytes: 100, + maxPayloadBytes: 100 + } + } +})); +vi.mock('node:worker_threads', () => ({ + parentPort: state.port, + workerData: state.data +})); +vi.mock('../../../demos/durable-jobs/thread-handler.mjs', () => ({ + get default() { + return state.handler; + } +})); +beforeEach(() => { + vi.resetModules(); + vi.clearAllMocks(); + state.handler = vi.fn(() => ({ url: '/result' })); + state.ackError = undefined; + state.data.moduleUrl = new URL( + '../../../demos/durable-jobs/thread-handler.mjs', + import.meta.url + ).href; + state.port.on.mockImplementation((_name, callback) => { + state.listener = callback; + }); + state.port.postMessage.mockImplementation(message => { + if (message.type === 'progress') + state.listener!({ + type: 'ack', + sequence: message.sequence, + error: state.ackError + }); + }); +}); + +it('waits for progress acknowledgements before sending the result', async () => { + state.handler = vi.fn(async (_input, context) => { + expect(context).toMatchObject({ runId: 'run', attempt: 2 }); + state.listener!({ type: 'ack', sequence: 999 }); + await context.report({ percent: 50 }); + return { url: '/result' }; + }); + await import('./thread-entry.js'); + expect( + state.port.postMessage.mock.calls.map(([message]) => message.type) + ).toEqual(['progress', 'result']); + expect(state.port.postMessage).toHaveBeenLastCalledWith({ + type: 'result', + value: { url: '/result' } + }); + expect(state.port.close).toHaveBeenCalledOnce(); +}); + +it('preserves permanent progress errors from the parent', async () => { + state.ackError = { message: 'lease expired', code: 'lease_lost' }; + state.handler = async (_input: any, context: any) => { + await context.report({ percent: 50 }); + return {}; + }; + await import('./thread-entry.js'); + expect(state.port.postMessage).toHaveBeenLastCalledWith({ + type: 'error', + error: state.ackError, + retryable: false + }); + expect(state.port.close).toHaveBeenCalledOnce(); +}); + +it('enforces progress count and payload limits, including forgotten promises', async () => { + state.handler = async (_input: any, context: any) => { + await context.report({ percent: 10 }); + await context.report({ percent: 20 }); + await context.report({ percent: 30 }); + return {}; + }; + await import('./thread-entry.js'); + expect(state.port.postMessage).toHaveBeenLastCalledWith( + expect.objectContaining({ + type: 'error', + retryable: false, + error: expect.objectContaining({ code: 'progress_limit' }) + }) + ); + expect( + state.port.postMessage.mock.calls.filter( + ([message]) => message.type === 'progress' + ) + ).toHaveLength(2); +}); + +it('reports invalid outputs and import failures and closes the port', async () => { + state.handler = () => ({ bad: undefined }); + await import('./thread-entry.js'); + expect(state.port.postMessage).toHaveBeenLastCalledWith( + expect.objectContaining({ + type: 'error', + retryable: false, + error: expect.objectContaining({ code: 'invalid_payload' }) + }) + ); + vi.resetModules(); + state.data.moduleUrl = 'file:///missing-handler.mjs'; + await import('./thread-entry.js'); + expect(state.port.postMessage).toHaveBeenLastCalledWith( + expect.objectContaining({ type: 'error', retryable: true }) + ); +}); + +it('requires a default function and exposes cancellation to handlers', async () => { + state.handler = undefined; + await import('./thread-entry.js'); + expect(state.port.postMessage).toHaveBeenLastCalledWith( + expect.objectContaining({ + error: expect.objectContaining({ code: 'invalid_handler' }), + retryable: false + }) + ); + vi.resetModules(); + state.handler = (_input: any, context: any) => { + state.listener!({ type: 'abort' }); + expect(context.signal.aborted).toBe(true); + return {}; + }; + await import('./thread-entry.js'); + expect(state.port.postMessage).toHaveBeenLastCalledWith({ + type: 'result', + value: {} + }); +}); diff --git a/libs/scheduler/src/worker.thread.test.ts b/libs/scheduler/src/worker.thread.test.ts new file mode 100644 index 00000000..a63d4320 --- /dev/null +++ b/libs/scheduler/src/worker.thread.test.ts @@ -0,0 +1,131 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { testJob } from '../testing/repository-contract.js'; +import { + InMemoryJobRepository, + JobScheduler, + type JobWorker +} from './index.js'; + +const { threads } = vi.hoisted(() => ({ threads: [] as any[] })); +vi.mock('node:worker_threads', async () => { + const { EventEmitter } = await import('node:events'); + return { + Worker: class extends EventEmitter { + stdout = { resume: vi.fn() }; + stderr = { resume: vi.fn() }; + terminate = vi.fn(async () => { + this.emit('exit', 0); + return 0; + }); + postMessage = vi.fn((message: any) => { + if (message.type === 'ack') { + if (message.error) + this.emit('message', { + type: 'error', + error: message.error, + retryable: false + }); + else + this.emit('message', { + type: 'result', + value: { url: '/result' } + }); + } + }); + constructor( + readonly url: URL, + readonly options: any + ) { + super(); + threads.push(this); + queueMicrotask(() => { + const mode = options.workerData.input.id; + if (mode === 'hang') return; + if (mode === 'error') { + this.emit('error', new Error('thread crashed')); + return; + } + if (mode === 'exit') { + this.emit('exit', 7); + return; + } + if (mode === 'retryable') { + this.emit('message', { + type: 'error', + error: { message: 'temporary' }, + retryable: true + }); + return; + } + this.emit('message', { type: 'ignored' }); + this.emit('message', { + type: 'progress', + sequence: 1, + data: { percent: mode === 'invalid' ? 'bad' : 50 } + }); + }); + } + } + }; +}); +const workers: JobWorker[] = []; +afterEach(async () => { + await Promise.all( + workers.splice(0).map(worker => worker.stop({ drainTimeoutMs: 30 })) + ); + threads.length = 0; +}); + +describe('worker thread coordination', () => { + it.each(['success', 'invalid', 'error', 'exit', 'hang', 'retryable'])( + 'settles %s and releases the worker', + async mode => { + const repository = new InMemoryJobRepository(); + const scheduler = new JobScheduler({ + storageRepository: repository + }); + const job = testJob({ timeoutMs: mode === 'hang' ? 30 : 1000 }); + const run = await scheduler.enqueue(job, { id: mode }); + const worker = scheduler.createWorker({ + pollIntervalMs: 2, + jobs: [job.thread(new URL('file:///jobs/handler.mjs'))] + }); + workers.push(worker); + await worker.start(); + await vi.waitFor(async () => + expect(await scheduler.getRun(job, run.id)).toHaveProperty( + 'status', + mode === 'success' ? 'succeeded' : 'failed' + ) + ); + const thread = threads[0]; + expect(thread.stdout.resume).toHaveBeenCalledOnce(); + expect(thread.stderr.resume).toHaveBeenCalledOnce(); + expect(thread.terminate).toHaveBeenCalledOnce(); + expect(thread.options.workerData.runId).toBe(run.id); + if (mode === 'success') { + expect(thread.postMessage).toHaveBeenCalledWith({ + type: 'ack', + sequence: 1 + }); + expect( + (await repository.events('default', run.id, 0)).map( + event => event.type + ) + ).toEqual(['queued', 'running', 'progress', 'succeeded']); + } + if (mode === 'hang') { + expect(thread.postMessage).toHaveBeenCalledWith({ + type: 'abort' + }); + expect(await scheduler.getRun(job, run.id)).toMatchObject({ + error: { code: 'timeout' } + }); + } + if (mode === 'invalid') + expect(await scheduler.getRun(job, run.id)).toMatchObject({ + error: { code: 'invalid_payload' } + }); + } + ); +}); diff --git a/libs/server-openapi/README.md b/libs/server-openapi/README.md index 5102b438..0d17676b 100644 --- a/libs/server-openapi/README.md +++ b/libs/server-openapi/README.md @@ -3,7 +3,7 @@ [![CI](https://github.com/cleverbrush/framework/actions/workflows/ci.yml/badge.svg)](https://github.com/cleverbrush/framework/actions/workflows/ci.yml) [![License: BSD-3-Clause](https://img.shields.io/badge/license-BSD--3--Clause-blue.svg)](../../LICENSE) -![Unit coverage](https://img.shields.io/badge/unit_coverage-96.4%25-brightgreen) +![Unit coverage](https://img.shields.io/badge/unit_coverage-96.7%25-brightgreen) OpenAPI 3.1 specification generation for [`@cleverbrush/server`](../server). Converts endpoint registrations, schema definitions, and authentication configuration into a fully-formed OpenAPI document — no annotations, no decorators. Also generates **AsyncAPI 3.0** documents for WebSocket subscription endpoints. diff --git a/libs/server/README.md b/libs/server/README.md index b13f3c80..8cb7fef2 100644 --- a/libs/server/README.md +++ b/libs/server/README.md @@ -3,7 +3,7 @@ [![CI](https://github.com/cleverbrush/framework/actions/workflows/ci.yml/badge.svg)](https://github.com/cleverbrush/framework/actions/workflows/ci.yml) [![License: BSD-3-Clause](https://img.shields.io/badge/license-BSD--3--Clause-blue.svg)](../../LICENSE) -![Unit coverage](https://img.shields.io/badge/unit_coverage-81.6%25-green) +![Unit coverage](https://img.shields.io/badge/unit_coverage-92%25-brightgreen) A schema-first HTTP server framework for Node.js. Combines [`@cleverbrush/schema`](../schema) for request validation, [`@cleverbrush/di`](../di) for dependency injection, and [`@cleverbrush/auth`](../auth) for authentication — all wired together through a fluent builder API. @@ -698,6 +698,9 @@ server.handle(GetUser, ({ params }) => { ## WebSocket Subscriptions +Middleware that rejects a subscription closes its WebSocket with code `1008`. +Closing or disconnecting aborts the handler signal and disposes its DI scope. + Define real-time endpoints using `endpoint.subscription()`: ```ts diff --git a/libs/server/src/Server.subscription.test.ts b/libs/server/src/Server.subscription.test.ts new file mode 100644 index 00000000..26bd5912 --- /dev/null +++ b/libs/server/src/Server.subscription.test.ts @@ -0,0 +1,165 @@ +import { once } from 'node:events'; +import { object, string } from '@cleverbrush/schema'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import WebSocket from 'ws'; +import { endpoint, mapHandlers } from './Endpoint.js'; +import { type Server, ServerBuilder } from './Server.js'; +import { tracked } from './Subscription.js'; + +const servers: Server[] = []; +const sockets: WebSocket[] = []; +afterEach(async () => { + for (const ws of sockets.splice(0)) ws.terminate(); + await Promise.all(servers.splice(0).map(s => s.close())); + vi.restoreAllMocks(); +}); +async function connect(def: any, handler: any, options?: any) { + const builder = new ServerBuilder(); + builder.handleAll( + mapHandlers( + { live: { events: def } }, + { live: { events: { handler, ...options } } } + ) + ); + const server = await builder.listen(0, '127.0.0.1'); + servers.push(server); + const ws = new WebSocket(`ws://127.0.0.1:${server.address!.port}/events`); + sockets.push(ws); + const messages: any[] = []; + ws.on('message', buffer => messages.push(JSON.parse(buffer.toString()))); + const closed = once(ws, 'close'); + await once(ws, 'open'); + return { ws, messages, closed }; +} +describe('subscription server lifecycle', () => { + it('validates incoming frames, supports ping, and streams tracked and ordinary events', async () => { + const definition = endpoint + .subscription('/events') + .incoming(object({ text: string() })) + .outgoing(string()); + const { ws, messages } = await connect( + definition, + async function* ({ incoming }) { + yield tracked('one', 'ready'); + for await (const message of incoming) { + yield message.text; + break; + } + } + ); + await vi.waitFor(() => + expect(messages).toContainEqual( + expect.objectContaining({ + type: 'tracked', + id: 'one', + data: 'ready' + }) + ) + ); + ws.send('invalid JSON'); + ws.send(JSON.stringify({ type: 'ping' })); + ws.send(JSON.stringify({ type: 'message', data: { text: 1 } })); + ws.send( + JSON.stringify({ type: 'message', data: { text: 'accepted' } }) + ); + await vi.waitFor(() => + expect(messages).toContainEqual( + expect.objectContaining({ type: 'message', data: 'accepted' }) + ) + ); + expect(messages).toContainEqual( + expect.objectContaining({ type: 'pong' }) + ); + expect(messages).toContainEqual( + expect.objectContaining({ type: 'error', code: 400 }) + ); + expect(messages).toContainEqual( + expect.objectContaining({ type: 'error', code: 422 }) + ); + }); + it('rejects invalid ordinary and tracked outputs without leaking handler errors', async () => { + vi.spyOn(console, 'error').mockImplementation(() => {}); + const { messages, closed } = await connect( + endpoint.subscription('/events').outgoing(string()), + async function* () { + yield 123; + yield tracked('invalid', 123); + throw new Error('private credentials'); + } + ); + expect((await closed)[0]).toBe(1011); + expect(messages.filter(m => m.type === 'error')).toHaveLength(3); + expect(JSON.stringify(messages)).not.toContain('private credentials'); + }); + it.each(['query', 'headers'])( + 'closes invalid %s contracts before calling handlers', + async kind => { + const handler = vi.fn(async function* () { + yield 'unexpected'; + }); + const def = endpoint + .subscription('/events') + [kind](object({ required: string() })); + const { closed } = await connect(def, handler); + expect((await closed)[0]).toBe(1002); + expect(handler).not.toHaveBeenCalled(); + } + ); + it('rejects failed middleware and isolates middleware exceptions', async () => { + for (const throws of [false, true]) { + const handler = vi.fn(async function* () { + yield 'unexpected'; + }); + const { closed } = await connect( + endpoint.subscription('/events'), + handler, + { + middlewares: [ + async ctx => { + if (throws) throw new Error('middleware failed'); + ctx.responded = true; + } + ] + } + ); + expect((await closed)[0]).toBe(throws ? 1011 : 1008); + expect(handler).not.toHaveBeenCalled(); + } + }); + it('ends pending incoming iteration and aborts the handler on disconnect', async () => { + let signal!: AbortSignal; + const finished = vi.fn(); + const { ws, messages, closed } = await connect( + endpoint.subscription('/events'), + async function* (ctx) { + signal = ctx.signal; + yield 'ready'; + for await (const message of ctx.incoming) yield message; + finished(); + } + ); + await vi.waitFor(() => expect(messages).toHaveLength(1)); + ws.close(); + await closed; + await vi.waitFor(() => expect(finished).toHaveBeenCalledOnce()); + expect(signal.aborted).toBe(true); + }); + it('limits unconsumed message queues', async () => { + const { ws, closed, messages } = await connect( + endpoint.subscription('/events'), + async function* ({ signal }) { + await new Promise(resolve => + signal.addEventListener('abort', () => resolve(), { + once: true + }) + ); + } + ); + for (let i = 0; i < 1100; i++) + ws.send(JSON.stringify({ type: 'message', data: i })); + expect((await closed)[0]).toBe(1008); + expect(messages).toContainEqual( + expect.objectContaining({ type: 'error', code: 429 }) + ); + }); +}); diff --git a/libs/server/src/Server.ts b/libs/server/src/Server.ts index abd5c4cb..2f53823b 100644 --- a/libs/server/src/Server.ts +++ b/libs/server/src/Server.ts @@ -1126,6 +1126,10 @@ export class Server { abortController ); }) + .then(() => { + // Short-circuiting middleware never invokes the terminal handler. + if (ctx.responded) ws.close(1008, 'Unauthorized'); + }) .catch(() => { ws.close(1011, 'Internal Server Error'); }); diff --git a/libs/server/src/multipart.limits.test.ts b/libs/server/src/multipart.limits.test.ts new file mode 100644 index 00000000..dd454f13 --- /dev/null +++ b/libs/server/src/multipart.limits.test.ts @@ -0,0 +1,123 @@ +import type { IncomingMessage } from 'node:http'; +import { PassThrough } from 'node:stream'; +import { array, object } from '@cleverbrush/schema'; +import { describe, expect, it } from 'vitest'; +import { parseMultipart } from './multipart.js'; +import { file, type UploadConfiguration } from './upload.js'; + +function field(name: string, value: string) { + return `--boundary\r\nContent-Disposition: form-data; name="${name}"\r\n\r\n${value}\r\n`; +} +function upload(name: string, value = 'data', mime = 'text/plain') { + return `--boundary\r\nContent-Disposition: form-data; name="${name}"; filename="a.txt"\r\nContent-Type: ${mime}\r\n\r\n${value}\r\n`; +} +async function parse( + parts: string[], + options: UploadConfiguration = {}, + limit = 4096, + headers = {} +) { + const stream = new PassThrough(); + const req = Object.assign(stream, { + headers: { + 'content-type': 'multipart/form-data; boundary=boundary', + ...headers + }, + complete: true + }) as unknown as IncomingMessage; + const result = parseMultipart(req, options, limit); + stream.end(parts.join('') + '--boundary--\r\n'); + try { + return await result; + } finally { + expect(req.listenerCount('aborted')).toBe(0); + stream.destroy(); + } +} +describe('multipart resource and contract boundaries', () => { + it('collects ordered file arrays, ordinary fields and empty required arrays', async () => { + const result = await parse( + [ + field('title', 'report'), + upload('attachments', 'one'), + upload('attachments', 'two') + ], + { + schema: object({ + attachments: array(file()), + missing: array(file()), + optional: file().optional() + }) + } + ); + expect(result.fields).toEqual({ title: 'report' }); + expect( + (result.files.attachments as any[]).map(f => f.buffer.toString()) + ).toEqual(['one', 'two']); + expect(result.files.missing).toEqual([]); + expect(result.files).not.toHaveProperty('optional'); + }); + it('returns rejected legacy files but rejects disallowed contract MIME types', async () => { + const result = await parse( + [upload('allowed'), upload('image', 'binary', 'image/png')], + { allowedMimeTypes: ['text/*'] } + ); + expect(result.files.allowed).toMatchObject({ size: 4 }); + expect(result.files).not.toHaveProperty('image'); + expect(result.rejectedFiles).toMatchObject([ + { fieldName: 'image', mimeType: 'image/png' } + ]); + await expect( + parse([upload('attachment')], { + schema: object({ attachment: file() }), + allowedMimeTypes: ['image/png'] + }) + ).rejects.toMatchObject({ status: 400 }); + }); + it.each([ + [[field('x', 'a'), field('x', 'b')], {}], + [[upload('x'), upload('x')], {}], + [[field('x', 'a'), upload('x')], {}], + [[upload('x'), field('x', 'a')], {}], + [[upload('unknown')], { schema: object({ expected: file() }) }], + [ + [field('expected', 'not a file')], + { schema: object({ expected: file() }) } + ], + [[], { schema: object({ expected: file() }) }] + ] as const)( + 'rejects duplicate, unknown and incorrectly encoded parts', + async (parts, options) => { + await expect(parse([...parts], options)).rejects.toMatchObject({ + status: 400 + }); + } + ); + it.each([ + [[upload('long/name~x')], { maxFieldNameSize: 2 }], + [[field('long', 'a')], { maxFieldNameSize: 2 }], + [[field('x', 'long')], { maxFieldSize: 2 }], + [[upload('x', 'long')], { maxFileSize: 2 }], + [[upload('x'), upload('y')], { maxFileCount: 1 }], + [[field('x', 'a'), field('y', 'b')], { maxFieldCount: 1 }], + [[field('x', 'a'), upload('y')], { maxPartCount: 1 }] + ] as const)('enforces multipart limits', async (parts, options) => { + await expect(parse([...parts], options)).rejects.toMatchObject({ + status: 413 + }); + }); + it('bounds declared and streamed wire sizes and escapes issue pointers', async () => { + await expect( + parse([], {}, 10, { 'content-length': '100' }) + ).rejects.toMatchObject({ status: 413 }); + await expect(parse([field('x', 'long')], {}, 10)).rejects.toMatchObject( + { status: 413 } + ); + await expect( + parse([upload('a/b~c')], { maxFieldNameSize: 1 }) + ).rejects.toMatchObject({ + status: 413, + extensions: { errors: [{ pointer: '/files/a~1b~0c' }] } + }); + }); +});