From a697ebe3401d5ad3afb7929a25f5e48a70488154 Mon Sep 17 00:00:00 2001 From: forehalo Date: Thu, 25 Apr 2024 03:03:45 +0000 Subject: [PATCH] feat(core): orm (#6536) --- .../common/infra/src/orm/affine/client.ts | 4 + packages/common/infra/src/orm/affine/hooks.ts | 21 ++ packages/common/infra/src/orm/affine/index.ts | 3 + .../common/infra/src/orm/affine/schema.ts | 17 ++ .../src/orm/core/__tests__/entity.spec.ts | 125 ++++++++++ .../infra/src/orm/core/__tests__/hook.spec.ts | 142 ++++++++++++ .../src/orm/core/__tests__/schema.spec.ts | 137 +++++++++++ .../infra/src/orm/core/__tests__/sync.spec.ts | 143 ++++++++++++ .../infra/src/orm/core/__tests__/yjs.spec.ts | 213 ++++++++++++++++++ .../infra/src/orm/core/adapters/index.ts | 4 + .../infra/src/orm/core/adapters/memory/db.ts | 17 ++ .../src/orm/core/adapters/memory/table.ts | 100 ++++++++ .../src/orm/core/adapters/mixins/hook.ts | 60 +++++ .../src/orm/core/adapters/mixins/index.ts | 1 + .../infra/src/orm/core/adapters/types.ts | 28 +++ .../infra/src/orm/core/adapters/yjs/db.ts | 44 ++++ .../infra/src/orm/core/adapters/yjs/table.ts | 193 ++++++++++++++++ packages/common/infra/src/orm/core/client.ts | 73 ++++++ packages/common/infra/src/orm/core/index.ts | 4 + packages/common/infra/src/orm/core/schema.ts | 55 +++++ packages/common/infra/src/orm/core/table.ts | 201 +++++++++++++++++ .../infra/src/orm/core/validators/data.ts | 142 ++++++++++++ .../infra/src/orm/core/validators/index.ts | 50 ++++ .../infra/src/orm/core/validators/schema.ts | 42 ++++ .../infra/src/orm/core/validators/types.ts | 10 + .../infra/src/orm/core/validators/yjs.ts | 35 +++ packages/common/infra/src/orm/index.ts | 1 + .../infra/src/sync/doc/__tests__/sync.spec.ts | 125 +--------- .../infra/src/sync/doc/__tests__/utils.ts | 108 +++++++++ 29 files changed, 1980 insertions(+), 118 deletions(-) create mode 100644 packages/common/infra/src/orm/affine/client.ts create mode 100644 packages/common/infra/src/orm/affine/hooks.ts create mode 100644 packages/common/infra/src/orm/affine/index.ts create mode 100644 packages/common/infra/src/orm/affine/schema.ts create mode 100644 packages/common/infra/src/orm/core/__tests__/entity.spec.ts create mode 100644 packages/common/infra/src/orm/core/__tests__/hook.spec.ts create mode 100644 packages/common/infra/src/orm/core/__tests__/schema.spec.ts create mode 100644 packages/common/infra/src/orm/core/__tests__/sync.spec.ts create mode 100644 packages/common/infra/src/orm/core/__tests__/yjs.spec.ts create mode 100644 packages/common/infra/src/orm/core/adapters/index.ts create mode 100644 packages/common/infra/src/orm/core/adapters/memory/db.ts create mode 100644 packages/common/infra/src/orm/core/adapters/memory/table.ts create mode 100644 packages/common/infra/src/orm/core/adapters/mixins/hook.ts create mode 100644 packages/common/infra/src/orm/core/adapters/mixins/index.ts create mode 100644 packages/common/infra/src/orm/core/adapters/types.ts create mode 100644 packages/common/infra/src/orm/core/adapters/yjs/db.ts create mode 100644 packages/common/infra/src/orm/core/adapters/yjs/table.ts create mode 100644 packages/common/infra/src/orm/core/client.ts create mode 100644 packages/common/infra/src/orm/core/index.ts create mode 100644 packages/common/infra/src/orm/core/schema.ts create mode 100644 packages/common/infra/src/orm/core/table.ts create mode 100644 packages/common/infra/src/orm/core/validators/data.ts create mode 100644 packages/common/infra/src/orm/core/validators/index.ts create mode 100644 packages/common/infra/src/orm/core/validators/schema.ts create mode 100644 packages/common/infra/src/orm/core/validators/types.ts create mode 100644 packages/common/infra/src/orm/core/validators/yjs.ts create mode 100644 packages/common/infra/src/orm/index.ts create mode 100644 packages/common/infra/src/sync/doc/__tests__/utils.ts diff --git a/packages/common/infra/src/orm/affine/client.ts b/packages/common/infra/src/orm/affine/client.ts new file mode 100644 index 000000000..fc77da83b --- /dev/null +++ b/packages/common/infra/src/orm/affine/client.ts @@ -0,0 +1,4 @@ +import { createORMClientType } from '../core'; +import { AFFiNE_DB_SCHEMA } from './schema'; + +export const ORMClient = createORMClientType(AFFiNE_DB_SCHEMA); diff --git a/packages/common/infra/src/orm/affine/hooks.ts b/packages/common/infra/src/orm/affine/hooks.ts new file mode 100644 index 000000000..727c3111f --- /dev/null +++ b/packages/common/infra/src/orm/affine/hooks.ts @@ -0,0 +1,21 @@ +import { ORMClient } from './client'; + +// The ORM hooks are used to define the transformers that will be applied on entities when they are loaded from the data providers. +// All transformers are doing in memory, none of the data under the hood will be changed. +// +// for example: +// data in providers: { color: 'red' } +// hook: { color: 'red' } => { color: '#FF0000' } +// +// ORMClient.defineHook( +// 'demo', +// 'deprecate color field and introduce colors filed', +// { +// deserialize(tag) { +// tag.color = stringToHex(tag.color) +// return tag; +// }, +// } +// ); + +export { ORMClient }; diff --git a/packages/common/infra/src/orm/affine/index.ts b/packages/common/infra/src/orm/affine/index.ts new file mode 100644 index 000000000..e0cd95422 --- /dev/null +++ b/packages/common/infra/src/orm/affine/index.ts @@ -0,0 +1,3 @@ +import './hooks'; + +export { ORMClient } from './client'; diff --git a/packages/common/infra/src/orm/affine/schema.ts b/packages/common/infra/src/orm/affine/schema.ts new file mode 100644 index 000000000..be2e6f903 --- /dev/null +++ b/packages/common/infra/src/orm/affine/schema.ts @@ -0,0 +1,17 @@ +import type { DBSchemaBuilder } from '../core'; +// import { f } from './core'; + +export const AFFiNE_DB_SCHEMA = { + // demo: { + // id: f.string().primaryKey().optional().default(nanoid), + // name: f.string(), + // // v1 + // // color: f.string(), + // // v2, without data level breaking change + // /** + // * @deprecated use [colors] + // */ + // color: f.string().optional(), // <= mark as optional since new created record might only have [colors] field + // colors: f.json().optional(), // <= mark as optional since old records might only have [color] field + // }, +} as const satisfies DBSchemaBuilder; diff --git a/packages/common/infra/src/orm/core/__tests__/entity.spec.ts b/packages/common/infra/src/orm/core/__tests__/entity.spec.ts new file mode 100644 index 000000000..bcb847822 --- /dev/null +++ b/packages/common/infra/src/orm/core/__tests__/entity.spec.ts @@ -0,0 +1,125 @@ +import { nanoid } from 'nanoid'; +import { + afterEach, + beforeEach, + describe, + expect, + test as t, + type TestAPI, +} from 'vitest'; + +import { + createORMClientType, + type DBSchemaBuilder, + f, + MemoryORMAdapter, + Table, +} from '../'; + +const TEST_SCHEMA = { + tags: { + id: f.string().primaryKey().default(nanoid), + name: f.string(), + color: f.string(), + }, +} satisfies DBSchemaBuilder; + +const Client = createORMClientType(TEST_SCHEMA); +type Context = { + client: InstanceType; +}; + +beforeEach(async t => { + t.client = new Client(new MemoryORMAdapter()); + await t.client.connect(); +}); + +afterEach(async t => { + await t.client.disconnect(); +}); + +const test = t as TestAPI; + +describe('ORM entity CRUD', () => { + test('should be able to create ORM client', t => { + const { client } = t; + + expect(client.tags instanceof Table).toBe(true); + }); + + test('should be able to create entity', async t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + expect(tag.id).toBeDefined(); + expect(tag.name).toBe('test'); + expect(tag.color).toBe('red'); + }); + + test('should be able to read entity', async t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + const tag2 = client.tags.get(tag.id); + expect(tag2).toEqual(tag); + }); + + test('should be able to list keys', t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + expect(client.tags.keys()).toStrictEqual([tag.id]); + + client.tags.delete(tag.id); + expect(client.tags.keys()).toStrictEqual([]); + }); + + test('should be able to update entity', async t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + client.tags.update(tag.id, { + name: 'test2', + }); + + const tag2 = client.tags.get(tag.id); + expect(tag2).toEqual({ + id: tag.id, + name: 'test2', + color: 'red', + }); + + // old tag should not be updated + expect(tag.name).not.toBe(tag2.name); + }); + + test('should be able to delete entity', async t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + client.tags.delete(tag.id); + + const tag2 = client.tags.get(tag.id); + expect(tag2).toBe(null); + }); +}); diff --git a/packages/common/infra/src/orm/core/__tests__/hook.spec.ts b/packages/common/infra/src/orm/core/__tests__/hook.spec.ts new file mode 100644 index 000000000..ca2618e4e --- /dev/null +++ b/packages/common/infra/src/orm/core/__tests__/hook.spec.ts @@ -0,0 +1,142 @@ +import { nanoid } from 'nanoid'; +import { + afterEach, + beforeEach, + describe, + expect, + test as t, + type TestAPI, +} from 'vitest'; + +import { + createORMClientType, + type DBSchemaBuilder, + type Entity, + f, + MemoryORMAdapter, +} from '../'; + +const TEST_SCHEMA = { + tags: { + id: f.string().primaryKey().default(nanoid), + name: f.string(), + color: f.string().optional(), + colors: f.json().optional(), + }, + badges: { + id: f.string().primaryKey().default(nanoid), + color: f.string(), + }, +} satisfies DBSchemaBuilder; + +const Client = createORMClientType(TEST_SCHEMA); + +// define the hooks +Client.defineHook('tags', 'migrate field `color` to field `colors`', { + deserialize(data) { + if (!data.colors && data.color) { + data.colors = [data.color]; + } + + return data; + }, +}); + +type Context = { + client: InstanceType; +}; + +beforeEach(async t => { + t.client = new Client(new MemoryORMAdapter()); + await t.client.connect(); +}); + +afterEach(async t => { + await t.client.disconnect(); +}); + +const test = t as TestAPI; + +describe('ORM hook mixin', () => { + test('create entity', t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + expect(tag.colors).toStrictEqual(['red']); + }); + + test('read entity', t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + const tag2 = client.tags.get(tag.id); + expect(tag2.colors).toStrictEqual(['red']); + }); + + test('update entity', t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + const tag2 = client.tags.update(tag.id, { color: 'blue' }); + expect(tag2.colors).toStrictEqual(['blue']); + }); + + test('subscribe entity', t => { + const { client } = t; + + let tag: Entity<(typeof TEST_SCHEMA)['tags']> | null = null; + const subscription = client.tags.get$('test').subscribe(data => { + tag = data; + }); + + client.tags.create({ + id: 'test', + name: 'test', + color: 'red', + }); + + expect(tag!.colors).toStrictEqual(['red']); + client.tags.update(tag!.id, { color: 'blue' }); + expect(tag!.colors).toStrictEqual(['blue']); + subscription.unsubscribe(); + }); + + test('should not run hook on unrelated entity', t => { + const { client } = t; + + const badge = client.badges.create({ + color: 'red', + }); + + // @ts-expect-error test + expect(badge.colors).toBeUndefined(); + }); + + test('should not touch the data in storage', t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + expect(tag.colors).toStrictEqual(['red']); + + // @ts-expect-error private + const rawTag = client.tags.adapter.data.get(tag.id); + expect(rawTag.color).toBe('red'); + expect(rawTag.colors).toBe(null); + }); +}); diff --git a/packages/common/infra/src/orm/core/__tests__/schema.spec.ts b/packages/common/infra/src/orm/core/__tests__/schema.spec.ts new file mode 100644 index 000000000..78189c1cb --- /dev/null +++ b/packages/common/infra/src/orm/core/__tests__/schema.spec.ts @@ -0,0 +1,137 @@ +import { nanoid } from 'nanoid'; +import { describe, expect, test } from 'vitest'; + +import { createORMClientType, f, MemoryORMAdapter } from '../'; + +describe('Schema validations', () => { + test('primary key must be set', () => { + expect(() => + createORMClientType({ + tags: { + id: f.string(), + name: f.string(), + }, + }) + ).toThrow( + '[Table(tags)]: There should be at least one field marked as primary key.' + ); + }); + + test('primary key must be unique', () => { + expect(() => + createORMClientType({ + tags: { + id: f.string().primaryKey(), + name: f.string().primaryKey(), + }, + }) + ).toThrow( + '[Table(tags)]: There should be only one field marked as primary key.' + ); + }); + + test('primary key should not be optional without default value', () => { + expect(() => + createORMClientType({ + tags: { + id: f.string().primaryKey().optional(), + name: f.string(), + }, + }) + ).toThrow( + "[Table(tags)]: Field 'id' can't be marked primary key and optional with no default value provider at the same time." + ); + }); + + test('primary key can be optional with default value', async () => { + expect(() => + createORMClientType({ + tags: { + id: f.string().primaryKey().optional().default(nanoid), + name: f.string(), + }, + }) + ).not.throws(); + }); +}); + +describe('Entity validations', () => { + const Client = createORMClientType({ + tags: { + id: f.string().primaryKey().default(nanoid), + name: f.string(), + color: f.string(), + }, + }); + + function createClient() { + return new Client(new MemoryORMAdapter()); + } + + test('should not update primary key', () => { + const client = createClient(); + + const tag = client.tags.create({ + name: 'tag', + color: 'blue', + }); + + // @ts-expect-error test + expect(() => client.tags.update(tag.id, { id: 'new-id' })).toThrow( + "[Table(tags)]: Primary key field 'id' can't be updated." + ); + }); + + test('should throw when trying to create entity with missing required field', () => { + const client = createClient(); + + // @ts-expect-error test + expect(() => client.tags.create({ name: 'test' })).toThrow( + "[Table(tags)]: Field 'color' is required but not set." + ); + }); + + test('should throw when trying to create entity with extra field', () => { + const client = createClient(); + + expect(() => + // @ts-expect-error test + client.tags.create({ name: 'test', color: 'red', extra: 'field' }) + ).toThrow("[Table(tags)]: Field 'extra' is not defined but set in entity."); + }); + + test('should throw when trying to create entity with unexpected field type', () => { + const client = createClient(); + + expect(() => + // @ts-expect-error test + client.tags.create({ name: 'test', color: 123 }) + ).toThrow( + "[Table(tags)]: Field 'color' type mismatch. Expected type 'string' but got 'number'." + ); + + expect(() => + // @ts-expect-error test + client.tags.create({ name: 'test', color: [123] }) + ).toThrow( + "[Table(tags)]: Field 'color' type mismatch. Expected type 'string' but got 'json'" + ); + }); + + test('should be able to assign `null` to json field', () => { + expect(() => { + const Client = createORMClientType({ + tags: { + id: f.string().primaryKey().default(nanoid), + info: f.json(), + }, + }); + + const client = new Client(new MemoryORMAdapter()); + + const tag = client.tags.create({ info: null }); + + expect(tag.info).toBe(null); + }); + }); +}); diff --git a/packages/common/infra/src/orm/core/__tests__/sync.spec.ts b/packages/common/infra/src/orm/core/__tests__/sync.spec.ts new file mode 100644 index 000000000..018ba0b89 --- /dev/null +++ b/packages/common/infra/src/orm/core/__tests__/sync.spec.ts @@ -0,0 +1,143 @@ +import { nanoid } from 'nanoid'; +import { + afterEach, + beforeEach, + describe, + expect, + test as t, + type TestAPI, + vitest, +} from 'vitest'; +import { Doc } from 'yjs'; + +import { DocEngine } from '../../../sync'; +import { MiniSyncServer } from '../../../sync/doc/__tests__/utils'; +import { MemoryStorage } from '../../../sync/doc/storage'; +import { + createORMClientType, + type DBSchemaBuilder, + f, + YjsDBAdapter, +} from '../'; + +const TEST_SCHEMA = { + tags: { + id: f.string().primaryKey().default(nanoid), + name: f.string(), + color: f.string().optional(), + colors: f.json().optional(), + }, +} satisfies DBSchemaBuilder; + +const Client = createORMClientType(TEST_SCHEMA); + +// define the hooks +Client.defineHook('tags', 'migrate field `color` to field `colors`', { + deserialize(data) { + if (!data.colors && data.color) { + data.colors = [data.color]; + } + + return data; + }, +}); + +type Context = { + server: MiniSyncServer; + user1: { + client: InstanceType; + engine: DocEngine; + }; + user2: { + client: InstanceType; + engine: DocEngine; + }; +}; + +function createEngine(server: MiniSyncServer) { + return new DocEngine(new MemoryStorage(), server.client()); +} + +async function createClient(server: MiniSyncServer, clientId: number) { + const engine = createEngine(server); + const client = new Client( + new YjsDBAdapter({ + getDoc(guid: string) { + const doc = new Doc({ guid }); + doc.clientID = clientId; + engine.addDoc(doc); + return doc; + }, + }) + ); + + return { + engine, + client, + }; +} + +beforeEach(async t => { + t.server = new MiniSyncServer(); + // we set user2's clientId greater than user1's clientId, + // so all conflicts will be resolved to user2's changes + t.user1 = await createClient(t.server, 1); + t.user2 = await createClient(t.server, 2); + + t.user1.engine.start(); + await t.user1.client.connect(); + t.user2.engine.start(); + await t.user2.client.connect(); +}); + +afterEach(async t => { + t.user1.client.disconnect(); + t.user2.client.disconnect(); + t.user1.engine.stop(); + t.user2.engine.stop(); +}); + +const test = t as TestAPI; + +describe('ORM compatibility in synchronization scenerio', () => { + test('2 clients create at the same time', async t => { + const { user1, user2 } = t; + const tag1 = user1.client.tags.create({ + name: 'tag1', + color: 'blue', + }); + + const tag2 = user2.client.tags.create({ + name: 'tag2', + color: 'red', + }); + + await vitest.waitFor(() => { + expect(user1.client.tags.keys()).toHaveLength(2); + expect(user2.client.tags.keys()).toHaveLength(2); + }); + + expect(user2.client.tags.get(tag1.id)).toStrictEqual(tag1); + expect(user1.client.tags.get(tag2.id)).toStrictEqual(tag2); + }); + + test('2 clients updating the same entity', async t => { + const { user1, user2 } = t; + const tag = user1.client.tags.create({ + name: 'tag1', + color: 'blue', + }); + + await vitest.waitFor(() => { + expect(user2.client.tags.keys()).toHaveLength(1); + }); + + user1.client.tags.update(tag.id, { color: 'red' }); + user2.client.tags.update(tag.id, { color: 'gray' }); + + await vitest.waitFor(() => { + expect(user1.client.tags.get(tag.id)).toHaveProperty('color', 'gray'); + expect(user2.client.tags.get(tag.id)).toHaveProperty('color', 'gray'); + }); + }); +}); diff --git a/packages/common/infra/src/orm/core/__tests__/yjs.spec.ts b/packages/common/infra/src/orm/core/__tests__/yjs.spec.ts new file mode 100644 index 000000000..b7aa05541 --- /dev/null +++ b/packages/common/infra/src/orm/core/__tests__/yjs.spec.ts @@ -0,0 +1,213 @@ +import { nanoid } from 'nanoid'; +import { + afterEach, + beforeEach, + describe, + expect, + test as t, + type TestAPI, +} from 'vitest'; +import { Doc } from 'yjs'; + +import { + createORMClientType, + type DBSchemaBuilder, + type DocProvider, + type Entity, + f, + Table, + YjsDBAdapter, +} from '../'; + +const TEST_SCHEMA = { + tags: { + id: f.string().primaryKey().default(nanoid), + name: f.string(), + color: f.string(), + }, +} satisfies DBSchemaBuilder; + +const docProvider: DocProvider = { + getDoc(guid: string) { + return new Doc({ guid }); + }, +}; + +const Client = createORMClientType(TEST_SCHEMA); +type Context = { + client: InstanceType; +}; + +beforeEach(async t => { + t.client = new Client(new YjsDBAdapter(docProvider)); + await t.client.connect(); +}); + +afterEach(async t => { + await t.client.disconnect(); +}); + +const test = t as TestAPI; + +describe('ORM entity CRUD', () => { + test('should be able to create ORM client', t => { + const { client } = t; + + expect(client.tags instanceof Table).toBe(true); + }); + + test('should be able to create entity', t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + expect(tag.id).toBeDefined(); + expect(tag.name).toBe('test'); + expect(tag.color).toBe('red'); + }); + + test('should be able to read entity', t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + const tag2 = client.tags.get(tag.id); + expect(tag2).toEqual(tag); + }); + + test('should be able to update entity', t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + client.tags.update(tag.id, { + name: 'test2', + }); + + const tag2 = client.tags.get(tag.id); + expect(tag2).toEqual({ + id: tag.id, + name: 'test2', + color: 'red', + }); + + // old tag should not be updated + expect(tag.name).not.toBe(tag2.name); + }); + + test('should be able to delete entity', t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + client.tags.delete(tag.id); + + const tag2 = client.tags.get(tag.id); + expect(tag2).toBe(null); + }); + + test('should be able to list keys', t => { + const { client } = t; + + const tag = client.tags.create({ + name: 'test', + color: 'red', + }); + + expect(client.tags.keys()).toStrictEqual([tag.id]); + + client.tags.delete(tag.id); + expect(client.tags.keys()).toStrictEqual([]); + }); + + test('should be able to subscribe to entity changes', t => { + const { client } = t; + + let tag: Entity<(typeof TEST_SCHEMA)['tags']> | null = null; + const subscription1 = client.tags.get$('test').subscribe(data => { + tag = data; + }); + + const subscription2 = client.tags.get$('test').subscribe(_ => {}); + + expect(tag).toBe(null); + + // create + client.tags.create({ + id: 'test', + name: 'testTag', + color: 'blue', + }); + + expect(tag!.id).toEqual('test'); + expect(tag!.color).toEqual('blue'); + + client.tags.update('test', { + color: 'red', + }); + expect(tag!.color).toEqual('red'); + + client.tags.delete('test'); + expect(tag).toBe(null); + + // internal status + subscription1.unsubscribe(); + // @ts-expect-error private field + expect(client.tags.subscribedKeys.size).toBe(1); + + subscription2.unsubscribe(); + // @ts-expect-error private field + expect(client.tags.subscribedKeys.size).toBe(0); + }); + + test('should be able to subscribe to entity key list', t => { + const { client } = t; + + let keys: string[] = []; + const subscription = client.tags.keys$().subscribe(data => { + keys = data; + }); + + client.tags.create({ + id: 'test', + name: 'testTag', + color: 'blue', + }); + + expect(keys).toStrictEqual(['test']); + + client.tags.update('test', { color: 'red' }); + expect(keys).toStrictEqual(['test']); + + client.tags.delete('test'); + expect(keys).toStrictEqual([]); + + subscription.unsubscribe(); + }); + + test('can not use reserved keyword as field name', () => { + const Client = createORMClientType({ + tags: { + $$KEY: f.string().primaryKey().default(nanoid), + }, + }); + + expect(() => + new Client(new YjsDBAdapter(docProvider)).connect() + ).rejects.toThrow( + "[Table(tags)]: Field '$$KEY' is reserved keyword and can't be used" + ); + }); +}); diff --git a/packages/common/infra/src/orm/core/adapters/index.ts b/packages/common/infra/src/orm/core/adapters/index.ts new file mode 100644 index 000000000..23e41b866 --- /dev/null +++ b/packages/common/infra/src/orm/core/adapters/index.ts @@ -0,0 +1,4 @@ +export * from './memory/db'; +export * from './mixins'; +export * from './types'; +export * from './yjs/db'; diff --git a/packages/common/infra/src/orm/core/adapters/memory/db.ts b/packages/common/infra/src/orm/core/adapters/memory/db.ts new file mode 100644 index 000000000..3de9abeb3 --- /dev/null +++ b/packages/common/infra/src/orm/core/adapters/memory/db.ts @@ -0,0 +1,17 @@ +import type { DBSchemaBuilder } from '../../schema'; +import type { DBAdapter } from '../types'; +import { MemoryTableAdapter } from './table'; + +export class MemoryORMAdapter implements DBAdapter { + connect(_db: DBSchemaBuilder): Promise { + return Promise.resolve(); + } + + disconnect(_db: DBSchemaBuilder): Promise { + return Promise.resolve(); + } + + table(tableName: string) { + return new MemoryTableAdapter(tableName); + } +} diff --git a/packages/common/infra/src/orm/core/adapters/memory/table.ts b/packages/common/infra/src/orm/core/adapters/memory/table.ts new file mode 100644 index 000000000..d7f677436 --- /dev/null +++ b/packages/common/infra/src/orm/core/adapters/memory/table.ts @@ -0,0 +1,100 @@ +import { merge } from 'lodash-es'; + +import { HookAdapter } from '../mixins'; +import type { Key, TableAdapter, TableOptions } from '../types'; + +@HookAdapter() +export class MemoryTableAdapter implements TableAdapter { + data = new Map(); + subscriptions = new Map void>>(); + + constructor(private readonly tableName: string) {} + + setup(_opts: TableOptions) {} + dispose() {} + + create(key: Key, data: any) { + if (this.data.has(key)) { + throw new Error( + `Record with key ${key} already exists in table ${this.tableName}` + ); + } + + this.data.set(key, data); + this.dispatch(key, data); + this.dispatch('$$KEYS', this.keys()); + return data; + } + + get(key: Key) { + return this.data.get(key) || null; + } + + subscribe(key: Key, callback: (data: any) => void): () => void { + const sKey = key.toString(); + let subs = this.subscriptions.get(sKey.toString()); + + if (!subs) { + subs = []; + this.subscriptions.set(sKey, subs); + } + + subs.push(callback); + callback(this.data.get(key) || null); + + return () => { + this.subscriptions.set( + sKey, + subs.filter(s => s !== callback) + ); + }; + } + + keys(): Key[] { + return Array.from(this.data.keys()); + } + + subscribeKeys(callback: (keys: Key[]) => void): () => void { + const sKey = `$$KEYS`; + let subs = this.subscriptions.get(sKey); + + if (!subs) { + subs = []; + this.subscriptions.set(sKey, subs); + } + subs.push(callback); + callback(this.keys()); + + return () => { + this.subscriptions.set( + sKey, + subs.filter(s => s !== callback) + ); + }; + } + + update(key: Key, data: any) { + let record = this.data.get(key); + + if (!record) { + throw new Error( + `Record with key ${key} does not exist in table ${this.tableName}` + ); + } + + record = merge({}, record, data); + this.data.set(key, record); + this.dispatch(key, record); + return record; + } + + delete(key: Key) { + this.data.delete(key); + this.dispatch(key, null); + this.dispatch('$$KEYS', this.keys()); + } + + dispatch(key: Key, data: any) { + this.subscriptions.get(key)?.forEach(callback => callback(data)); + } +} diff --git a/packages/common/infra/src/orm/core/adapters/mixins/hook.ts b/packages/common/infra/src/orm/core/adapters/mixins/hook.ts new file mode 100644 index 000000000..2cdf9dd64 --- /dev/null +++ b/packages/common/infra/src/orm/core/adapters/mixins/hook.ts @@ -0,0 +1,60 @@ +import type { Key, TableAdapter, TableOptions } from '../types'; + +declare module '../types' { + interface TableOptions { + hooks?: Hook[]; + } +} + +export interface Hook { + deserialize(dbVal: T): T; +} + +export interface TableAdapterWithHook extends Hook {} + +export function HookAdapter(): ClassDecorator { + // @ts-expect-error allow + return (Class: { new (...args: any[]): TableAdapter }) => { + return class TableAdapterImpl + extends Class + implements TableAdapterWithHook + { + hooks: Hook[] = []; + + deserialize(data: unknown) { + if (!this.hooks.length) { + return data; + } + + return this.hooks.reduce( + (acc, hook) => hook.deserialize(acc), + Object.assign({} as any, data) + ); + } + + override setup(opts: TableOptions) { + this.hooks = opts.hooks || []; + super.setup(opts); + } + + override create(key: Key, data: any) { + return this.deserialize(super.create(key, data)); + } + + override get(key: Key) { + return this.deserialize(super.get(key)); + } + + override update(key: Key, data: any) { + return this.deserialize(super.update(key, data)); + } + + override subscribe( + key: Key, + callback: (data: unknown) => void + ): () => void { + return super.subscribe(key, data => callback(this.deserialize(data))); + } + }; + }; +} diff --git a/packages/common/infra/src/orm/core/adapters/mixins/index.ts b/packages/common/infra/src/orm/core/adapters/mixins/index.ts new file mode 100644 index 000000000..1146a6d5c --- /dev/null +++ b/packages/common/infra/src/orm/core/adapters/mixins/index.ts @@ -0,0 +1 @@ +export * from './hook'; diff --git a/packages/common/infra/src/orm/core/adapters/types.ts b/packages/common/infra/src/orm/core/adapters/types.ts new file mode 100644 index 000000000..6d274826e --- /dev/null +++ b/packages/common/infra/src/orm/core/adapters/types.ts @@ -0,0 +1,28 @@ +import type { DBSchemaBuilder, TableSchemaBuilder } from '../schema'; + +export interface Key { + toString(): string; +} + +export interface TableOptions { + schema: TableSchemaBuilder; +} + +export interface TableAdapter { + setup(opts: TableOptions): void; + dispose(): void; + create(key: K, data: Partial): T; + get(key: K): T; + subscribe(key: K, callback: (data: T) => void): () => void; + keys(): K[]; + subscribeKeys(callback: (keys: K[]) => void): () => void; + update(key: K, data: Partial): T; + delete(key: K): void; +} + +export interface DBAdapter { + connect(db: DBSchemaBuilder): Promise; + disconnect(db: DBSchemaBuilder): Promise; + + table(tableName: string): TableAdapter; +} diff --git a/packages/common/infra/src/orm/core/adapters/yjs/db.ts b/packages/common/infra/src/orm/core/adapters/yjs/db.ts new file mode 100644 index 000000000..aad83144f --- /dev/null +++ b/packages/common/infra/src/orm/core/adapters/yjs/db.ts @@ -0,0 +1,44 @@ +import type { Doc } from 'yjs'; + +import type { DBSchemaBuilder } from '../../schema'; +import { validators } from '../../validators'; +import type { DBAdapter, TableAdapter } from '../types'; +import { YjsTableAdapter } from './table'; + +export interface DocProvider { + getDoc(guid: string): Doc; +} + +export class YjsDBAdapter implements DBAdapter { + tables: Map = new Map(); + constructor(private readonly provider: DocProvider) {} + + connect(db: DBSchemaBuilder): Promise { + for (const [tableName, table] of Object.entries(db)) { + validators.validateYjsTableSchema(tableName, table); + const doc = this.provider.getDoc(tableName); + + this.tables.set(tableName, new YjsTableAdapter(tableName, doc)); + } + + return Promise.resolve(); + } + + disconnect(_db: DBSchemaBuilder): Promise { + this.tables.forEach(table => { + table.dispose(); + }); + this.tables.clear(); + return Promise.resolve(); + } + + table(tableName: string) { + const table = this.tables.get(tableName); + + if (!table) { + throw new Error('Table not found'); + } + + return table; + } +} diff --git a/packages/common/infra/src/orm/core/adapters/yjs/table.ts b/packages/common/infra/src/orm/core/adapters/yjs/table.ts new file mode 100644 index 000000000..7948a0c2a --- /dev/null +++ b/packages/common/infra/src/orm/core/adapters/yjs/table.ts @@ -0,0 +1,193 @@ +import { omit } from 'lodash-es'; +import type { Doc, Map as YMap, Transaction, YMapEvent } from 'yjs'; + +import { validators } from '../../validators'; +import { HookAdapter } from '../mixins'; +import type { Key, TableAdapter, TableOptions } from '../types'; + +/** + * Yjs Adapter for AFFiNE ORM + * + * Structure: + * + * Each table is a YDoc instance + * + * Table(YDoc) + * Key(string): Row(YMap)({ + * FieldA(string): Value(Primitive) + * FieldB(string): Value(Primitive) + * FieldC(string): Value(Primitive) + * }) + */ +@HookAdapter() +export class YjsTableAdapter implements TableAdapter { + private readonly deleteFlagKey = '$$DELETED'; + private readonly keyFlagKey = '$$KEY'; + private readonly hiddenFields = [this.deleteFlagKey, this.keyFlagKey]; + + private readonly origin = 'YjsTableAdapter'; + + keysCache: Set | null = null; + cacheStaled = true; + + constructor( + private readonly tableName: string, + private readonly doc: Doc + ) {} + + setup(_opts: TableOptions): void { + this.doc.on('update', (_, origin) => { + if (origin !== this.origin) { + this.markCacheStaled(); + } + }); + } + + dispose() { + this.doc.destroy(); + } + + create(key: Key, data: any) { + validators.validateYjsEntityData(this.tableName, data); + const record = this.doc.getMap(key.toString()); + + this.doc.transact(() => { + for (const key in data) { + record.set(key, data[key]); + } + + this.keyBy(record, key); + }, this.origin); + + this.markCacheStaled(); + return this.value(record); + } + + update(key: Key, data: any) { + validators.validateYjsEntityData(this.tableName, data); + const record = this.record(key); + + if (this.isDeleted(record)) { + return; + } + + this.doc.transact(() => { + for (const key in data) { + record.set(key, data[key]); + } + }, this.origin); + + return this.value(record); + } + + get(key: Key) { + const record = this.record(key); + return this.value(record); + } + + subscribe(key: Key, callback: (data: any) => void) { + const record: YMap = this.record(key); + // init callback + callback(this.value(record)); + + const ob = (event: YMapEvent) => { + callback(this.value(event.target)); + }; + record.observe(ob); + + return () => { + record.unobserve(ob); + }; + } + + keys() { + const keysCache = this.buildKeysCache(); + return Array.from(keysCache); + } + + subscribeKeys(callback: (keys: Key[]) => void) { + const keysCache = this.buildKeysCache(); + // init callback + callback(Array.from(keysCache)); + + const ob = (tx: Transaction) => { + const keysCache = this.buildKeysCache(); + + for (const [type] of tx.changed) { + const data = type as unknown as YMap; + const key = this.keyof(data); + if (this.isDeleted(data)) { + keysCache.delete(key); + } else { + keysCache.add(key); + } + } + + callback(Array.from(keysCache)); + }; + + this.doc.on('afterTransaction', ob); + + return () => { + this.doc.off('afterTransaction', ob); + }; + } + + delete(key: Key) { + const record = this.record(key); + + this.doc.transact(() => { + for (const key of record.keys()) { + if (!this.hiddenFields.includes(key)) { + record.delete(key); + } + } + record.set(this.deleteFlagKey, true); + }, this.origin); + this.markCacheStaled(); + } + + private isDeleted(record: YMap) { + return record.has(this.deleteFlagKey); + } + + private record(key: Key) { + return this.doc.getMap(key.toString()); + } + + private value(record: YMap) { + if (this.isDeleted(record) || !record.size) { + return null; + } + + return omit(record.toJSON(), this.hiddenFields); + } + + private buildKeysCache() { + if (!this.keysCache || this.cacheStaled) { + this.keysCache = new Set(); + + for (const key of this.doc.share.keys()) { + const record = this.doc.getMap(key); + if (!this.isDeleted(record)) { + this.keysCache.add(this.keyof(record)); + } + } + this.cacheStaled = false; + } + + return this.keysCache; + } + + private markCacheStaled() { + this.cacheStaled = true; + } + + private keyof(record: YMap) { + return record.get(this.keyFlagKey); + } + + private keyBy(record: YMap, key: Key) { + record.set(this.keyFlagKey, key); + } +} diff --git a/packages/common/infra/src/orm/core/client.ts b/packages/common/infra/src/orm/core/client.ts new file mode 100644 index 000000000..280c37b13 --- /dev/null +++ b/packages/common/infra/src/orm/core/client.ts @@ -0,0 +1,73 @@ +import { type DBAdapter, type Hook } from './adapters'; +import type { DBSchemaBuilder } from './schema'; +import { type CreateEntityInput, Table, type TableMap } from './table'; +import { validators } from './validators'; + +export class ORMClient { + static hooksMap: Map[]> = new Map(); + private readonly tables = new Map>(); + constructor( + protected readonly db: DBSchemaBuilder, + protected readonly adapter: DBAdapter + ) { + Object.entries(db).forEach(([tableName, tableSchema]) => { + Object.defineProperty(this, tableName, { + get: () => { + let table = this.tables.get(tableName); + if (!table) { + table = new Table(this.adapter, tableName, { + schema: tableSchema, + hooks: ORMClient.hooksMap.get(tableName), + }); + this.tables.set(tableName, table); + } + return table; + }, + }); + }); + } + + static defineHook(tableName: string, _desc: string, hook: Hook) { + let hooks = this.hooksMap.get(tableName); + if (!hooks) { + hooks = []; + this.hooksMap.set(tableName, hooks); + } + + hooks.push(hook); + } + + async connect() { + await this.adapter.connect(this.db); + } + + async disconnect() { + await this.adapter.disconnect(this.db); + } +} + +export function createORMClientType( + db: Schema +) { + Object.entries(db).forEach(([tableName, schema]) => { + validators.validateTableSchema(tableName, schema); + }); + + class ORMClientWithTables extends ORMClient { + constructor(adapter: DBAdapter) { + super(db, adapter); + } + } + + return ORMClientWithTables as { + new ( + ...args: ConstructorParameters + ): ORMClient & TableMap; + + defineHook( + tableName: TableName, + desc: string, + hook: Hook> + ): void; + }; +} diff --git a/packages/common/infra/src/orm/core/index.ts b/packages/common/infra/src/orm/core/index.ts new file mode 100644 index 000000000..323139e54 --- /dev/null +++ b/packages/common/infra/src/orm/core/index.ts @@ -0,0 +1,4 @@ +export * from './adapters'; +export * from './client'; +export * from './schema'; +export * from './table'; diff --git a/packages/common/infra/src/orm/core/schema.ts b/packages/common/infra/src/orm/core/schema.ts new file mode 100644 index 000000000..83f39f67c --- /dev/null +++ b/packages/common/infra/src/orm/core/schema.ts @@ -0,0 +1,55 @@ +export type FieldType = 'string' | 'number' | 'boolean' | 'json'; + +export interface FieldSchema { + type: FieldType; + optional: boolean; + isPrimaryKey: boolean; + default?: () => Type; +} + +export type TableSchema = Record; +export type TableSchemaBuilder = Record< + string, + FieldSchemaBuilder +>; +export type DBSchemaBuilder = Record; + +export class FieldSchemaBuilder< + Type = unknown, + Optional extends boolean = false, + PrimaryKey extends boolean = false, +> { + schema: FieldSchema = { + type: 'string', + optional: false, + isPrimaryKey: false, + default: undefined, + }; + + constructor(type: FieldType) { + this.schema.type = type; + } + + optional() { + this.schema.optional = true; + return this as FieldSchemaBuilder; + } + + default(value: () => Type) { + this.schema.default = value; + this.schema.optional = true; + return this as FieldSchemaBuilder; + } + + primaryKey() { + this.schema.isPrimaryKey = true; + return this as FieldSchemaBuilder; + } +} + +export const f = { + string: () => new FieldSchemaBuilder('string'), + number: () => new FieldSchemaBuilder('number'), + boolean: () => new FieldSchemaBuilder('boolean'), + json: () => new FieldSchemaBuilder('json'), +} satisfies Record FieldSchemaBuilder>; diff --git a/packages/common/infra/src/orm/core/table.ts b/packages/common/infra/src/orm/core/table.ts new file mode 100644 index 000000000..a45943b06 --- /dev/null +++ b/packages/common/infra/src/orm/core/table.ts @@ -0,0 +1,201 @@ +import { isUndefined, omitBy } from 'lodash-es'; +import { Observable, shareReplay } from 'rxjs'; + +import type { DBAdapter, Key, TableAdapter, TableOptions } from './adapters'; +import type { + DBSchemaBuilder, + FieldSchemaBuilder, + TableSchema, + TableSchemaBuilder, +} from './schema'; +import { validators } from './validators'; + +type Pretty = T extends any + ? { + -readonly [P in keyof T]: T[P]; + } + : never; + +type RequiredFields = { + [K in keyof T as T[K] extends FieldSchemaBuilder + ? Optional extends false + ? K + : never + : never]: T[K] extends FieldSchemaBuilder ? Type : never; +}; + +type OptionalFields = { + [K in keyof T as T[K] extends FieldSchemaBuilder + ? Optional extends true + ? K + : never + : never]?: T[K] extends FieldSchemaBuilder ? Type : never; +}; + +type PrimaryKeyField = { + [K in keyof T]: T[K] extends FieldSchemaBuilder + ? PrimaryKey extends true + ? K + : never + : never; +}[keyof T]; + +export type NonPrimaryKeyFields = { + [K in keyof T]: T[K] extends FieldSchemaBuilder + ? PrimaryKey extends false + ? K + : never + : never; +}[keyof T]; + +export type PrimaryKeyFieldType = + T[PrimaryKeyField] extends FieldSchemaBuilder + ? Type extends Key + ? Type + : never + : never; + +export type CreateEntityInput = Pretty< + RequiredFields & OptionalFields +>; + +// @TODO(@forehalo): return value need to be specified with `Default` inference +export type Entity = Pretty< + CreateEntityInput & { + [key in PrimaryKeyField]: PrimaryKeyFieldType; + } +>; + +export type UpdateEntityInput = Pretty<{ + [key in NonPrimaryKeyFields]?: T[key] extends FieldSchemaBuilder< + infer Type + > + ? Type + : never; +}>; + +export class Table { + readonly schema: TableSchema; + readonly keyField: string = ''; + private readonly adapter: TableAdapter, Entity>; + + private readonly subscribedKeys: Map> = new Map(); + + constructor( + db: DBAdapter, + public readonly name: string, + private readonly opts: TableOptions + ) { + this.adapter = db.table(name) as any; + this.adapter.setup(opts); + this.schema = Object.entries(this.opts.schema).reduce( + (acc, [fieldName, fieldBuilder]) => { + acc[fieldName] = fieldBuilder.schema; + if (fieldBuilder.schema.isPrimaryKey) { + // @ts-expect-error still in constructor + this.keyField = fieldName; + } + return acc; + }, + {} as TableSchema + ); + } + + create(input: CreateEntityInput): Entity { + const data = Object.entries(this.schema).reduce( + (acc, [key, schema]) => { + const inputVal = acc[key]; + + if (inputVal === undefined) { + if (schema.optional) { + acc[key] = null; + } + + if (schema.default) { + acc[key] = schema.default() ?? null; + } + } + + return acc; + }, + omitBy(input, isUndefined) as any + ); + + validators.validateCreateEntityData(this, data); + + return this.adapter.create(data[this.keyField], data); + } + + update(key: PrimaryKeyFieldType, input: UpdateEntityInput): Entity { + validators.validateUpdateEntityData(this, input); + return this.adapter.update(key, omitBy(input, isUndefined) as any); + } + + get(key: PrimaryKeyFieldType): Entity { + return this.adapter.get(key); + } + + get$(key: PrimaryKeyFieldType): Observable> { + let ob$ = this.subscribedKeys.get(key); + + if (!ob$) { + ob$ = new Observable>(subscriber => { + const unsubscribe = this.adapter.subscribe(key, data => { + subscriber.next(data); + }); + + return () => { + unsubscribe(); + this.subscribedKeys.delete(key); + }; + }).pipe( + shareReplay({ + refCount: true, + bufferSize: 1, + }) + ); + + this.subscribedKeys.set(key, ob$); + } + + return ob$; + } + + keys(): PrimaryKeyFieldType[] { + return this.adapter.keys(); + } + + keys$(): Observable[]> { + let ob$ = this.subscribedKeys.get('$$KEYS'); + + if (!ob$) { + ob$ = new Observable[]>(subscriber => { + const unsubscribe = this.adapter.subscribeKeys(keys => { + subscriber.next(keys); + }); + + return () => { + unsubscribe(); + this.subscribedKeys.delete('$$KEYS'); + }; + }).pipe( + shareReplay({ + refCount: true, + bufferSize: 1, + }) + ); + + this.subscribedKeys.set('$$KEYS', ob$); + } + + return ob$; + } + + delete(key: PrimaryKeyFieldType) { + return this.adapter.delete(key); + } +} + +export type TableMap = { + readonly [K in keyof Tables]: Table; +}; diff --git a/packages/common/infra/src/orm/core/validators/data.ts b/packages/common/infra/src/orm/core/validators/data.ts new file mode 100644 index 000000000..db176c117 --- /dev/null +++ b/packages/common/infra/src/orm/core/validators/data.ts @@ -0,0 +1,142 @@ +import { pick as lodashPick } from 'lodash-es'; + +import type { FieldType } from '../schema'; +import type { DataValidator } from './types'; + +function inputType(val: any) { + return val === null || + Array.isArray(val) || + val.constructor === 'Object' || + !val.constructor /* Object.create(null) */ + ? 'json' + : typeof val; +} + +function typeMatches(typeWant: FieldType, typeGet: string) { + if (typeWant === 'json') { + switch (typeGet) { + case 'bigint': + case 'function': + case 'object': // we've already converted available types into 'json' + case 'symbol': + case 'undefined': + return false; + } + } + + return typeWant === typeGet; +} + +export const dataValidators = { + PrimaryKeyShouldExist: { + validate(table, data) { + const val = data[table.keyField]; + + if (val === undefined || val === null) { + throw new Error( + `[Table(${table.name})]: Primary key field '${table.keyField}' is required but not set.` + ); + } + }, + }, + PrimaryKeyShouldNotBeUpdated: { + validate(table, data) { + if (data[table.keyField] !== undefined) { + throw new Error( + `[Table(${table.name})]: Primary key field '${table.keyField}' can't be updated.` + ); + } + }, + }, + DataTypeShouldMatch: { + validate(table, data) { + for (const key in data) { + const field = table.schema[key]; + if (!field) { + throw new Error( + `[Table(${table.name})]: Field '${key}' is not defined but set in entity.` + ); + } + + const val = data[key]; + + if (val === undefined) { + delete data[key]; + continue; + } + + if ( + val === null && + (!field.optional || + field.optional) /* say 'null' can be stored as 'json' */ + ) { + throw new Error( + `[Table(${table.name})]: Field '${key}' is required but set as null.` + ); + } + + const typeGet = inputType(val); + if (!typeMatches(field.type, typeGet)) { + throw new Error( + `[Table(${table.name})]: Field '${key}' type mismatch. Expected ${field.type} got ${typeGet}.` + ); + } + } + }, + }, + DataTypeShouldExactlyMatch: { + validate(table, data) { + const keys: Set = new Set(); + for (const key in data) { + const field = table.schema[key]; + if (!field) { + throw new Error( + `[Table(${table.name})]: Field '${key}' is not defined but set in entity.` + ); + } + + const val = data[key]; + + if ((val === undefined || val === null) && !field.optional) { + throw new Error( + `[Table(${table.name})]: Field '${key}' is required but not set.` + ); + } + + const typeGet = inputType(val); + if (!typeMatches(field.type, typeGet)) { + throw new Error( + `[Table(${table.name})]: Field '${key}' type mismatch. Expected type '${field.type}' but got '${typeGet}'.` + ); + } + + keys.add(key); + } + + for (const key in table.schema) { + if (!keys.has(key) && table.schema[key].optional === false) { + throw new Error( + `[Table(${table.name})]: Field '${key}' is required but not set.` + ); + } + } + }, + }, +} satisfies Record; + +// lodash pick's signature is not typesafe +const pick = lodashPick as >( + obj: T, + ...keys: Array +) => Pick; + +export const createEntityDataValidators = pick( + dataValidators, + 'PrimaryKeyShouldExist', + 'DataTypeShouldExactlyMatch' +); +export const updateEntityDataValidators = pick( + dataValidators, + 'PrimaryKeyShouldNotBeUpdated', + 'DataTypeShouldMatch' +); diff --git a/packages/common/infra/src/orm/core/validators/index.ts b/packages/common/infra/src/orm/core/validators/index.ts new file mode 100644 index 000000000..2af0002da --- /dev/null +++ b/packages/common/infra/src/orm/core/validators/index.ts @@ -0,0 +1,50 @@ +import { createEntityDataValidators, updateEntityDataValidators } from './data'; +import { tableSchemaValidators } from './schema'; +import { yjsDataValidators, yjsTableSchemaValidators } from './yjs'; + +interface ValidationError { + code: string; + error: Error; +} + +function throwIfError(errors: ValidationError[]) { + if (errors.length) { + const message = errors + .map(({ code, error }) => `${code}: ${error.stack ?? error.message}`) + .join('\n'); + + throw new Error('Validation Failed Error\n' + message); + } +} + +function validate void }>( + rules: Record, + ...payload: Parameters +) { + const errors: ValidationError[] = []; + + for (const [code, validator] of Object.entries(rules)) { + try { + validator.validate(...payload); + } catch (e) { + errors.push({ code, error: e as Error }); + } + } + + throwIfError(errors); +} + +function use void }>( + rules: Record +) { + return (...payload: Parameters) => + validate(rules, ...payload); +} + +export const validators = { + validateTableSchema: use(tableSchemaValidators), + validateCreateEntityData: use(createEntityDataValidators), + validateUpdateEntityData: use(updateEntityDataValidators), + validateYjsTableSchema: use(yjsTableSchemaValidators), + validateYjsEntityData: use(yjsDataValidators), +}; diff --git a/packages/common/infra/src/orm/core/validators/schema.ts b/packages/common/infra/src/orm/core/validators/schema.ts new file mode 100644 index 000000000..dd8c0fbe8 --- /dev/null +++ b/packages/common/infra/src/orm/core/validators/schema.ts @@ -0,0 +1,42 @@ +import type { TableSchemaValidator } from './types'; + +export const tableSchemaValidators: Record = { + PrimaryKeyShouldExist: { + validate(tableName, table) { + if (!Object.values(table).some(field => field.schema.isPrimaryKey)) { + throw new Error( + `[Table(${tableName})]: There should be at least one field marked as primary key.` + ); + } + }, + }, + OnlyOnePrimaryKey: { + validate(tableName, table) { + const primaryFields = []; + + for (const name in table) { + if (table[name].schema.isPrimaryKey) { + primaryFields.push(name); + } + } + + if (primaryFields.length > 1) { + throw new Error( + `[Table(${tableName})]: There should be only one field marked as primary key. Found [${primaryFields.join(', ')}].` + ); + } + }, + }, + PrimaryKeyShouldNotBeOptional: { + validate(tableName, table) { + for (const name in table) { + const opts = table[name].schema; + if (opts.isPrimaryKey && opts.optional && !opts.default) { + throw new Error( + `[Table(${tableName})]: Field '${name}' can't be marked primary key and optional with no default value provider at the same time.` + ); + } + } + }, + }, +}; diff --git a/packages/common/infra/src/orm/core/validators/types.ts b/packages/common/infra/src/orm/core/validators/types.ts new file mode 100644 index 000000000..a73d5c1c3 --- /dev/null +++ b/packages/common/infra/src/orm/core/validators/types.ts @@ -0,0 +1,10 @@ +import type { TableSchemaBuilder } from '../schema'; +import type { Table } from '../table'; + +export interface TableSchemaValidator { + validate(tableName: string, schema: TableSchemaBuilder): void; +} + +export interface DataValidator { + validate(table: Table, data: any): void; +} diff --git a/packages/common/infra/src/orm/core/validators/yjs.ts b/packages/common/infra/src/orm/core/validators/yjs.ts new file mode 100644 index 000000000..e78845947 --- /dev/null +++ b/packages/common/infra/src/orm/core/validators/yjs.ts @@ -0,0 +1,35 @@ +import type { TableSchemaValidator } from './types'; + +const PRESERVED_FIELDS = ['$$KEY', '$$DELETED']; + +interface DataValidator { + validate(tableName: string, data: any): void; +} + +export const yjsTableSchemaValidators: Record = { + UsePreservedFields: { + validate(tableName, table) { + for (const name in table) { + if (PRESERVED_FIELDS.includes(name)) { + throw new Error( + `[Table(${tableName})]: Field '${name}' is reserved keyword and can't be used.` + ); + } + } + }, + }, +}; + +export const yjsDataValidators: Record = { + SetPreservedFields: { + validate(tableName, data) { + for (const name of PRESERVED_FIELDS) { + if (data[name] !== undefined) { + throw new Error( + `[Table(${tableName})]: Field '${name}' is reserved keyword and can't be set.` + ); + } + } + }, + }, +}; diff --git a/packages/common/infra/src/orm/index.ts b/packages/common/infra/src/orm/index.ts new file mode 100644 index 000000000..8f9dc079c --- /dev/null +++ b/packages/common/infra/src/orm/index.ts @@ -0,0 +1 @@ +export * from './affine'; diff --git a/packages/common/infra/src/sync/doc/__tests__/sync.spec.ts b/packages/common/infra/src/sync/doc/__tests__/sync.spec.ts index b3d24aed5..437b21fd8 100644 --- a/packages/common/infra/src/sync/doc/__tests__/sync.spec.ts +++ b/packages/common/infra/src/sync/doc/__tests__/sync.spec.ts @@ -1,125 +1,14 @@ -import { nanoid } from 'nanoid'; import { describe, expect, test, vitest } from 'vitest'; -import { - diffUpdate, - Doc as YDoc, - encodeStateAsUpdate, - encodeStateVectorFromUpdate, - mergeUpdates, -} from 'yjs'; +import { Doc as YDoc, encodeStateAsUpdate } from 'yjs'; -import { AsyncLock } from '../../../utils'; import { DocEngine } from '..'; -import type { DocServer } from '../server'; import { MemoryStorage } from '../storage'; -import { isEmptyUpdate } from '../utils'; - -class MiniServer { - lock = new AsyncLock(); - db = new Map(); - listeners = new Set<{ - cb: (updates: { - docId: string; - data: Uint8Array; - serverClock: number; - }) => void; - clientId: string; - }>(); - - client() { - return new MiniServerClient(nanoid(), this); - } -} - -class MiniServerClient implements DocServer { - constructor( - private readonly id: string, - private readonly server: MiniServer - ) {} - - async pullDoc(docId: string, stateVector: Uint8Array) { - using _lock = await this.server.lock.acquire(); - const doc = this.server.db.get(docId); - if (!doc) { - return null; - } - const data = doc.data; - return { - data: - !isEmptyUpdate(data) && stateVector.length > 0 - ? diffUpdate(data, stateVector) - : data, - serverClock: 0, - stateVector: !isEmptyUpdate(data) - ? encodeStateVectorFromUpdate(data) - : new Uint8Array(), - }; - } - - async pushDoc( - docId: string, - data: Uint8Array - ): Promise<{ serverClock: number }> { - using _lock = await this.server.lock.acquire(); - const doc = this.server.db.get(docId); - const oldData = doc?.data ?? new Uint8Array(); - const newClock = (doc?.clock ?? 0) + 1; - this.server.db.set(docId, { - data: !isEmptyUpdate(data) - ? !isEmptyUpdate(oldData) - ? mergeUpdates([oldData, data]) - : data - : oldData, - clock: newClock, - }); - for (const { clientId, cb } of this.server.listeners) { - if (clientId !== this.id) { - cb({ - docId, - data, - serverClock: newClock, - }); - } - } - return { serverClock: newClock }; - } - - async loadServerClock(after: number): Promise> { - using _lock = await this.server.lock.acquire(); - const map = new Map(); - - for (const [docId, { clock }] of this.server.db) { - if (clock > after) { - map.set(docId, clock); - } - } - - return map; - } - - async subscribeAllDocs( - cb: (updates: { - docId: string; - data: Uint8Array; - serverClock: number; - }) => void - ): Promise<() => void> { - const listener = { cb, clientId: this.id }; - this.server.listeners.add(listener); - return () => { - this.server.listeners.delete(listener); - }; - } - - async waitForConnectingServer(): Promise {} - disconnectServer(): void {} - onInterrupted(_cb: (reason: string) => void): void {} -} +import { MiniSyncServer } from './utils'; describe('sync', () => { test('basic sync', async () => { const storage = new MemoryStorage(); - const server = new MiniServer(); + const server = new MiniSyncServer(); const engine = new DocEngine(storage, server.client()).start(); const doc = new YDoc({ guid: 'a' }); engine.addDoc(doc); @@ -132,7 +21,7 @@ describe('sync', () => { }); test('can pull from server', async () => { - const server = new MiniServer(); + const server = new MiniSyncServer(); { const engine = new DocEngine( new MemoryStorage(), @@ -158,7 +47,7 @@ describe('sync', () => { }); test('2 client', async () => { - const server = new MiniServer(); + const server = new MiniSyncServer(); await Promise.all([ (async () => { const engine = new DocEngine( @@ -190,7 +79,7 @@ describe('sync', () => { }); test('2 client share storage and eventBus (simulate different tabs in same browser)', async () => { - const server = new MiniServer(); + const server = new MiniSyncServer(); const storage = new MemoryStorage(); await Promise.all([ @@ -215,7 +104,7 @@ describe('sync', () => { }); test('legacy data', async () => { - const server = new MiniServer(); + const server = new MiniSyncServer(); const storage = new MemoryStorage(); { diff --git a/packages/common/infra/src/sync/doc/__tests__/utils.ts b/packages/common/infra/src/sync/doc/__tests__/utils.ts new file mode 100644 index 000000000..fc1ffeee2 --- /dev/null +++ b/packages/common/infra/src/sync/doc/__tests__/utils.ts @@ -0,0 +1,108 @@ +import { nanoid } from 'nanoid'; +import { diffUpdate, encodeStateVectorFromUpdate, mergeUpdates } from 'yjs'; + +import { AsyncLock } from '../../../utils'; +import type { DocServer } from '../server'; +import { isEmptyUpdate } from '../utils'; + +export class MiniSyncServer { + lock = new AsyncLock(); + db = new Map(); + listeners = new Set<{ + cb: (updates: { + docId: string; + data: Uint8Array; + serverClock: number; + }) => void; + clientId: string; + }>(); + + client() { + return new MiniServerClient(nanoid(), this); + } +} + +export class MiniServerClient implements DocServer { + constructor( + private readonly id: string, + private readonly server: MiniSyncServer + ) {} + + async pullDoc(docId: string, stateVector: Uint8Array) { + using _lock = await this.server.lock.acquire(); + const doc = this.server.db.get(docId); + if (!doc) { + return null; + } + const data = doc.data; + return { + data: + !isEmptyUpdate(data) && stateVector.length > 0 + ? diffUpdate(data, stateVector) + : data, + serverClock: 0, + stateVector: !isEmptyUpdate(data) + ? encodeStateVectorFromUpdate(data) + : new Uint8Array(), + }; + } + + async pushDoc( + docId: string, + data: Uint8Array + ): Promise<{ serverClock: number }> { + using _lock = await this.server.lock.acquire(); + const doc = this.server.db.get(docId); + const oldData = doc?.data ?? new Uint8Array(); + const newClock = (doc?.clock ?? 0) + 1; + this.server.db.set(docId, { + data: !isEmptyUpdate(data) + ? !isEmptyUpdate(oldData) + ? mergeUpdates([oldData, data]) + : data + : oldData, + clock: newClock, + }); + for (const { clientId, cb } of this.server.listeners) { + if (clientId !== this.id) { + cb({ + docId, + data, + serverClock: newClock, + }); + } + } + return { serverClock: newClock }; + } + + async loadServerClock(after: number): Promise> { + using _lock = await this.server.lock.acquire(); + const map = new Map(); + + for (const [docId, { clock }] of this.server.db) { + if (clock > after) { + map.set(docId, clock); + } + } + + return map; + } + + async subscribeAllDocs( + cb: (updates: { + docId: string; + data: Uint8Array; + serverClock: number; + }) => void + ): Promise<() => void> { + const listener = { cb, clientId: this.id }; + this.server.listeners.add(listener); + return () => { + this.server.listeners.delete(listener); + }; + } + + async waitForConnectingServer(): Promise {} + disconnectServer(): void {} + onInterrupted(_cb: (reason: string) => void): void {} +}