feat(core): orm (#6536)

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);

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 };

packages/common/infra/src/orm/affine/index.ts

@@ -0,0 +1,3 @@
+import './hooks';
+
+export { ORMClient } from './client';

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<string[]>().optional(), // <= mark as optional since old records might only have [color] field
+  // },
+} as const satisfies DBSchemaBuilder;

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<typeof Client>;
+};
+
+beforeEach<Context>(async t => {
+  t.client = new Client(new MemoryORMAdapter());
+  await t.client.connect();
+});
+
+afterEach<Context>(async t => {
+  await t.client.disconnect();
+});
+
+const test = t as TestAPI<Context>;
+
+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);
+  });
+});

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<string[]>().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<typeof Client>;
+};
+
+beforeEach<Context>(async t => {
+  t.client = new Client(new MemoryORMAdapter());
+  await t.client.connect();
+});
+
+afterEach<Context>(async t => {
+  await t.client.disconnect();
+});
+
+const test = t as TestAPI<Context>;
+
+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);
+  });
+});

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);
+    });
+  });
+});

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<string[]>().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<typeof Client>;
+    engine: DocEngine;
+  };
+  user2: {
+    client: InstanceType<typeof Client>;
+    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<Context>(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<Context>(async t => {
+  t.user1.client.disconnect();
+  t.user2.client.disconnect();
+  t.user1.engine.stop();
+  t.user2.engine.stop();
+});
+
+const test = t as TestAPI<Context>;
+
+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');
+    });
+  });
+});

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<typeof Client>;
+};
+
+beforeEach<Context>(async t => {
+  t.client = new Client(new YjsDBAdapter(docProvider));
+  await t.client.connect();
+});
+
+afterEach<Context>(async t => {
+  await t.client.disconnect();
+});
+
+const test = t as TestAPI<Context>;
+
+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"
+    );
+  });
+});

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';

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<void> {
+    return Promise.resolve();
+  }
+
+  disconnect(_db: DBSchemaBuilder): Promise<void> {
+    return Promise.resolve();
+  }
+
+  table(tableName: string) {
+    return new MemoryTableAdapter(tableName);
+  }
+}

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<Key, any>();
+  subscriptions = new Map<Key, Array<(data: any) => 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));
+  }
+}

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<unknown>[];
+  }
+}
+
+export interface Hook<T> {
+  deserialize(dbVal: T): T;
+}
+
+export interface TableAdapterWithHook<T = unknown> extends Hook<T> {}
+
+export function HookAdapter(): ClassDecorator {
+  // @ts-expect-error allow
+  return (Class: { new (...args: any[]): TableAdapter }) => {
+    return class TableAdapterImpl
+      extends Class
+      implements TableAdapterWithHook
+    {
+      hooks: Hook<unknown>[] = [];
+
+      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)));
+      }
+    };
+  };
+}

packages/common/infra/src/orm/core/adapters/mixins/index.ts

@@ -0,0 +1 @@
+export * from './hook';

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<K extends Key = any, T = unknown> {
+  setup(opts: TableOptions): void;
+  dispose(): void;
+  create(key: K, data: Partial<T>): 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>): T;
+  delete(key: K): void;
+}
+
+export interface DBAdapter {
+  connect(db: DBSchemaBuilder): Promise<void>;
+  disconnect(db: DBSchemaBuilder): Promise<void>;
+
+  table(tableName: string): TableAdapter;
+}

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<string, TableAdapter> = new Map();
+  constructor(private readonly provider: DocProvider) {}
+
+  connect(db: DBSchemaBuilder): Promise<void> {
+    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<void> {
+    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;
+  }
+}

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<Key> | 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<any> = this.record(key);
+    // init callback
+    callback(this.value(record));
+
+    const ob = (event: YMapEvent<any>) => {
+      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<any>;
+        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<any>) {
+    return record.has(this.deleteFlagKey);
+  }
+
+  private record(key: Key) {
+    return this.doc.getMap(key.toString());
+  }
+
+  private value(record: YMap<any>) {
+    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<any>) {
+    return record.get(this.keyFlagKey);
+  }
+
+  private keyBy(record: YMap<any>, key: Key) {
+    record.set(this.keyFlagKey, key);
+  }
+}

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<string, Hook<any>[]> = new Map();
+  private readonly tables = new Map<string, Table<any>>();
+  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<any>) {
+    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<Schema extends DBSchemaBuilder>(
+  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<typeof ORMClientWithTables>
+    ): ORMClient & TableMap<Schema>;
+
+    defineHook<TableName extends keyof Schema>(
+      tableName: TableName,
+      desc: string,
+      hook: Hook<CreateEntityInput<Schema[TableName]>>
+    ): void;
+  };
+}

packages/common/infra/src/orm/core/index.ts

@@ -0,0 +1,4 @@
+export * from './adapters';
+export * from './client';
+export * from './schema';
+export * from './table';

packages/common/infra/src/orm/core/schema.ts

@@ -0,0 +1,55 @@
+export type FieldType = 'string' | 'number' | 'boolean' | 'json';
+
+export interface FieldSchema<Type = unknown> {
+  type: FieldType;
+  optional: boolean;
+  isPrimaryKey: boolean;
+  default?: () => Type;
+}
+
+export type TableSchema = Record<string, FieldSchema>;
+export type TableSchemaBuilder = Record<
+  string,
+  FieldSchemaBuilder<any, boolean>
+>;
+export type DBSchemaBuilder = Record<string, TableSchemaBuilder>;
+
+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<Type, true, PrimaryKey>;
+  }
+
+  default(value: () => Type) {
+    this.schema.default = value;
+    this.schema.optional = true;
+    return this as FieldSchemaBuilder<Type, true, PrimaryKey>;
+  }
+
+  primaryKey() {
+    this.schema.isPrimaryKey = true;
+    return this as FieldSchemaBuilder<Type, Optional, true>;
+  }
+}
+
+export const f = {
+  string: () => new FieldSchemaBuilder<string>('string'),
+  number: () => new FieldSchemaBuilder<number>('number'),
+  boolean: () => new FieldSchemaBuilder<boolean>('boolean'),
+  json: <T = any>() => new FieldSchemaBuilder<T>('json'),
+} satisfies Record<FieldType, () => FieldSchemaBuilder<any>>;

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> = T extends any
+  ? {
+      -readonly [P in keyof T]: T[P];
+    }
+  : never;
+
+type RequiredFields<T extends TableSchemaBuilder> = {
+  [K in keyof T as T[K] extends FieldSchemaBuilder<any, infer Optional>
+    ? Optional extends false
+      ? K
+      : never
+    : never]: T[K] extends FieldSchemaBuilder<infer Type> ? Type : never;
+};
+
+type OptionalFields<T extends TableSchemaBuilder> = {
+  [K in keyof T as T[K] extends FieldSchemaBuilder<any, infer Optional>
+    ? Optional extends true
+      ? K
+      : never
+    : never]?: T[K] extends FieldSchemaBuilder<infer Type> ? Type : never;
+};
+
+type PrimaryKeyField<T extends TableSchemaBuilder> = {
+  [K in keyof T]: T[K] extends FieldSchemaBuilder<any, any, infer PrimaryKey>
+    ? PrimaryKey extends true
+      ? K
+      : never
+    : never;
+}[keyof T];
+
+export type NonPrimaryKeyFields<T extends TableSchemaBuilder> = {
+  [K in keyof T]: T[K] extends FieldSchemaBuilder<any, any, infer PrimaryKey>
+    ? PrimaryKey extends false
+      ? K
+      : never
+    : never;
+}[keyof T];
+
+export type PrimaryKeyFieldType<T extends TableSchemaBuilder> =
+  T[PrimaryKeyField<T>] extends FieldSchemaBuilder<infer Type>
+    ? Type extends Key
+      ? Type
+      : never
+    : never;
+
+export type CreateEntityInput<T extends TableSchemaBuilder> = Pretty<
+  RequiredFields<T> & OptionalFields<T>
+>;
+
+// @TODO(@forehalo): return value need to be specified with `Default` inference
+export type Entity<T extends TableSchemaBuilder> = Pretty<
+  CreateEntityInput<T> & {
+    [key in PrimaryKeyField<T>]: PrimaryKeyFieldType<T>;
+  }
+>;
+
+export type UpdateEntityInput<T extends TableSchemaBuilder> = Pretty<{
+  [key in NonPrimaryKeyFields<T>]?: T[key] extends FieldSchemaBuilder<
+    infer Type
+  >
+    ? Type
+    : never;
+}>;
+
+export class Table<T extends TableSchemaBuilder> {
+  readonly schema: TableSchema;
+  readonly keyField: string = '';
+  private readonly adapter: TableAdapter<PrimaryKeyFieldType<T>, Entity<T>>;
+
+  private readonly subscribedKeys: Map<Key, Observable<any>> = 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<T>): Entity<T> {
+    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<T>, input: UpdateEntityInput<T>): Entity<T> {
+    validators.validateUpdateEntityData(this, input);
+    return this.adapter.update(key, omitBy(input, isUndefined) as any);
+  }
+
+  get(key: PrimaryKeyFieldType<T>): Entity<T> {
+    return this.adapter.get(key);
+  }
+
+  get$(key: PrimaryKeyFieldType<T>): Observable<Entity<T>> {
+    let ob$ = this.subscribedKeys.get(key);
+
+    if (!ob$) {
+      ob$ = new Observable<Entity<T>>(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<T>[] {
+    return this.adapter.keys();
+  }
+
+  keys$(): Observable<PrimaryKeyFieldType<T>[]> {
+    let ob$ = this.subscribedKeys.get('$$KEYS');
+
+    if (!ob$) {
+      ob$ = new Observable<PrimaryKeyFieldType<T>[]>(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<T>) {
+    return this.adapter.delete(key);
+  }
+}
+
+export type TableMap<Tables extends DBSchemaBuilder> = {
+  readonly [K in keyof Tables]: Table<Tables[K]>;
+};

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<string> = 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<string, DataValidator>;
+
+// lodash pick's signature is not typesafe
+const pick = lodashPick as <T extends Record<string, any>>(
+  obj: T,
+  ...keys: Array<keyof T>
+) => Pick<T, keyof T>;
+
+export const createEntityDataValidators = pick(
+  dataValidators,
+  'PrimaryKeyShouldExist',
+  'DataTypeShouldExactlyMatch'
+);
+export const updateEntityDataValidators = pick(
+  dataValidators,
+  'PrimaryKeyShouldNotBeUpdated',
+  'DataTypeShouldMatch'
+);

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<Validator extends { validate: (...args: any[]) => void }>(
+  rules: Record<string, Validator>,
+  ...payload: Parameters<Validator['validate']>
+) {
+  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<Validator extends { validate: (...args: any[]) => void }>(
+  rules: Record<string, Validator>
+) {
+  return (...payload: Parameters<Validator['validate']>) =>
+    validate(rules, ...payload);
+}
+
+export const validators = {
+  validateTableSchema: use(tableSchemaValidators),
+  validateCreateEntityData: use(createEntityDataValidators),
+  validateUpdateEntityData: use(updateEntityDataValidators),
+  validateYjsTableSchema: use(yjsTableSchemaValidators),
+  validateYjsEntityData: use(yjsDataValidators),
+};

packages/common/infra/src/orm/core/validators/schema.ts

@@ -0,0 +1,42 @@
+import type { TableSchemaValidator } from './types';
+
+export const tableSchemaValidators: Record<string, TableSchemaValidator> = {
+  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.`
+          );
+        }
+      }
+    },
+  },
+};

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<any>, data: any): void;
+}

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<string, TableSchemaValidator> = {
+  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<string, DataValidator> = {
+  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.`
+          );
+        }
+      }
+    },
+  },
+};

packages/common/infra/src/orm/index.ts

@@ -0,0 +1 @@
+export * from './affine';

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 {
+import { Doc as YDoc, encodeStateAsUpdate } from 'yjs';
-  diffUpdate,
-  Doc as YDoc,
-  encodeStateAsUpdate,
-  encodeStateVectorFromUpdate,
-  mergeUpdates,
-} from 'yjs';
 
-import { AsyncLock } from '../../../utils';
 import { DocEngine } from '..';
-import type { DocServer } from '../server';
 import { MemoryStorage } from '../storage';
-import { isEmptyUpdate } from '../utils';
+import { MiniSyncServer } from './utils';
-
-class MiniServer {
-  lock = new AsyncLock();
-  db = new Map<string, { data: Uint8Array; clock: number }>();
-  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<Map<string, number>> {
-    using _lock = await this.server.lock.acquire();
-    const map = new Map<string, number>();
-
-    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<void> {}
-  disconnectServer(): void {}
-  onInterrupted(_cb: (reason: string) => void): void {}
-}
 
 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();
 
     {

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<string, { data: Uint8Array; clock: number }>();
+  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<Map<string, number>> {
+    using _lock = await this.server.lock.acquire();
+    const map = new Map<string, number>();
+
+    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<void> {}
+  disconnectServer(): void {}
+  onInterrupted(_cb: (reason: string) => void): void {}
+}