[Content Management] Server side client (#175968)

packages/kbn-object-versioning/lib/content_management_services_versioning.ts

@@ -69,18 +69,20 @@ const validateServiceDefinitions = (definitions: ServiceDefinitionVersioned) =>
  * ```ts
  * From this
  * {
+ *   // Service definition version 1
  *   1: {
  *     get: {
  *       in: {
- *         options: { up: () => {} } // 1
+ *         options: { up: () => {} }
  *       }
  *     },
  *     ...
  *   },
+ *   // Service definition version 2
  *   2: {
  *     get: {
  *       in: {
- *         options: { up: () => {} } // 2
+ *         options: { up: () => {} }
  *       }
  *     },
  *   }

src/plugins/content_management/common/rpc/types.ts

@@ -15,6 +15,7 @@ export interface ProcedureSchemas {
 export type ItemResult<T = unknown, M = void> = M extends void
   ? {
       item: T;
+      meta?: never;
     }
   : {
       item: T;

src/plugins/content_management/server/content_client/content_client.test.ts

@@ -0,0 +1,264 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+import { ContentCrud } from '../core/crud';
+import { EventBus } from '../core/event_bus';
+import { createMemoryStorage, type FooContent } from '../core/mocks';
+import { ContentClient } from './content_client';
+
+describe('ContentClient', () => {
+  const setup = ({
+    contentTypeId = 'foo',
+  }: {
+    contentTypeId?: string;
+  } = {}) => {
+    const storage = createMemoryStorage();
+    const eventBus = new EventBus();
+    const crudInstance = new ContentCrud<FooContent>(contentTypeId, storage, { eventBus });
+
+    const contentClient = ContentClient.create(contentTypeId, {
+      crudInstance,
+      storageContext: {} as any,
+    });
+
+    return { contentClient };
+  };
+
+  describe('instance', () => {
+    test('should throw an Error if instantiate using constructor', () => {
+      const expectToThrow = () => {
+        new ContentClient(Symbol('foo'), 'foo', {} as any);
+      };
+      expect(expectToThrow).toThrowError('Use ContentClient.create() instead');
+    });
+
+    test('should have contentTypeId', () => {
+      const { contentClient } = setup({ contentTypeId: 'hellooo' });
+      expect(contentClient.contentTypeId).toBe('hellooo');
+    });
+
+    test('should throw if crudInstance is not an instance of ContentCrud', () => {
+      const expectToThrow = () => {
+        ContentClient.create('foo', {
+          crudInstance: {} as any,
+          storageContext: {} as any,
+        });
+      };
+      // With this test and runtime check we can rely on all the existing tests of the Content Crud.
+      // e.g. the tests about events being dispatched, etc.
+      expect(expectToThrow).toThrowError('Crud instance missing or not an instance of ContentCrud');
+    });
+  });
+
+  describe('Crud', () => {
+    describe('create()', () => {
+      test('should create an item', async () => {
+        const { contentClient } = setup();
+        const itemCreated = await contentClient.create({ foo: 'bar' });
+        const { id } = itemCreated.result.item;
+        const res = await contentClient.get(id);
+        expect(res.result.item).toEqual({ foo: 'bar', id });
+      });
+
+      test('should pass the options to the storage', async () => {
+        const { contentClient } = setup();
+
+        const options = { forwardInResponse: { option1: 'foo' } };
+        const res = await contentClient.create({ field1: 123 }, options);
+        expect(res.result.item).toEqual({
+          field1: 123,
+          id: expect.any(String),
+          options: { option1: 'foo' }, // the options have correctly been passed to the storage
+        });
+      });
+    });
+
+    describe('get()', () => {
+      // Note: we test the client get() method in multiple other tests for
+      // the "create()" and "update()" methods, no need for extended tests here.
+      test('should return undefined if no item was found', async () => {
+        const { contentClient } = setup();
+        const res = await contentClient.get('hello');
+        expect(res.result.item).toBeUndefined();
+      });
+
+      test('should pass the options to the storage', async () => {
+        const { contentClient } = setup();
+
+        const options = { forwardInResponse: { foo: 'bar' } };
+        const res = await contentClient.get('hello', options);
+
+        expect(res.result.item).toEqual({
+          // the options have correctly been passed to the storage
+          options: { foo: 'bar' },
+        });
+      });
+    });
+
+    describe('bulkGet()', () => {
+      test('should return multiple items', async () => {
+        const { contentClient } = setup();
+
+        const item1 = await contentClient.create({ name: 'item1' });
+        const item2 = await contentClient.create({ name: 'item2' });
+        const ids = [item1.result.item.id, item2.result.item.id];
+
+        const res = await contentClient.bulkGet(ids);
+        expect(res.result.hits).toEqual([
+          {
+            item: {
+              name: 'item1',
+              id: expect.any(String),
+            },
+          },
+          {
+            item: {
+              name: 'item2',
+              id: expect.any(String),
+            },
+          },
+        ]);
+      });
+
+      test('should pass the options to the storage', async () => {
+        const { contentClient } = setup();
+
+        const item1 = await contentClient.create({ name: 'item1' });
+        const item2 = await contentClient.create({ name: 'item2' });
+        const ids = [item1.result.item.id, item2.result.item.id];
+
+        const options = { forwardInResponse: { foo: 'bar' } };
+        const res = await contentClient.bulkGet(ids, options);
+
+        expect(res.result.hits).toEqual([
+          {
+            item: {
+              name: 'item1',
+              id: expect.any(String),
+              options: { foo: 'bar' }, // the options have correctly been passed to the storage
+            },
+          },
+          {
+            item: {
+              name: 'item2',
+              id: expect.any(String),
+              options: { foo: 'bar' }, // the options have correctly been passed to the storage
+            },
+          },
+        ]);
+      });
+    });
+
+    describe('update()', () => {
+      test('should update an item', async () => {
+        const { contentClient } = setup();
+        const itemCreated = await contentClient.create({ foo: 'bar' });
+        const { id } = itemCreated.result.item;
+
+        await contentClient.update(id, { foo: 'changed' });
+
+        const res = await contentClient.get(id);
+        expect(res.result.item).toEqual({ foo: 'changed', id });
+      });
+
+      test('should pass the options to the storage', async () => {
+        const { contentClient } = setup();
+        const itemCreated = await contentClient.create({ field1: 'bar' });
+        const { id } = itemCreated.result.item;
+
+        const options = { forwardInResponse: { option1: 'foo' } };
+        const res = await contentClient.update(id, { field1: 'changed' }, options);
+
+        expect(res.result.item).toEqual({
+          field1: 'changed',
+          id,
+          options: { option1: 'foo' }, // the options have correctly been passed to the storage
+        });
+      });
+    });
+
+    describe('delete()', () => {
+      test('should delete an item', async () => {
+        const { contentClient } = setup();
+        const itemCreated = await contentClient.create({ foo: 'bar' });
+        const { id } = itemCreated.result.item;
+
+        {
+          const res = await contentClient.get(id);
+          expect(res.result.item).not.toBeUndefined();
+        }
+
+        await contentClient.delete(id);
+
+        {
+          const res = await contentClient.get(id);
+          expect(res.result.item).toBeUndefined();
+        }
+      });
+
+      test('should pass the options to the storage', async () => {
+        const { contentClient } = setup();
+        const itemCreated = await contentClient.create({ field1: 'bar' });
+        const { id } = itemCreated.result.item;
+
+        const options = { forwardInResponse: { option1: 'foo' } };
+
+        const res = await contentClient.delete(id, options);
+
+        expect(res.result).toEqual({
+          success: true,
+          options: { option1: 'foo' }, // the options have correctly been passed to the storage
+        });
+      });
+    });
+
+    describe('search()', () => {
+      test('should find an item', async () => {
+        const { contentClient } = setup();
+
+        await contentClient.create({ title: 'hello' });
+
+        const res = await contentClient.search({ text: 'hello' });
+
+        expect(res.result).toEqual({
+          hits: [
+            {
+              id: expect.any(String),
+              title: 'hello',
+            },
+          ],
+          pagination: {
+            cursor: '',
+            total: 1,
+          },
+        });
+      });
+
+      test('should pass the options to the storage', async () => {
+        const { contentClient } = setup();
+        await contentClient.create({ title: 'hello' });
+
+        const options = { forwardInResponse: { option1: 'foo' } };
+        const res = await contentClient.search({ text: 'hello' }, options);
+
+        expect(res.result).toEqual({
+          hits: [
+            {
+              id: expect.any(String),
+              title: 'hello',
+              options: { option1: 'foo' }, // the options have correctly been passed to the storage
+            },
+          ],
+          pagination: {
+            cursor: '',
+            total: 1,
+          },
+        });
+      });
+    });
+  });
+});

src/plugins/content_management/server/content_client/content_client.ts

@@ -0,0 +1,57 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+import type { StorageContext } from '../core';
+import { ContentCrud } from '../core/crud';
+import type { IContentClient } from './types';
+
+interface Context<T = unknown> {
+  crudInstance: ContentCrud<T>;
+  storageContext: StorageContext;
+}
+
+const secretToken = Symbol('secretToken');
+
+export class ContentClient<T = unknown> implements IContentClient<T> {
+  static create<T = unknown>(contentTypeId: string, ctx: Context<T>): IContentClient<T> {
+    return new ContentClient<T>(secretToken, contentTypeId, ctx);
+  }
+
+  constructor(token: symbol, public contentTypeId: string, private readonly ctx: Context<T>) {
+    if (token !== secretToken) {
+      throw new Error('Use ContentClient.create() instead');
+    }
+
+    if (ctx.crudInstance instanceof ContentCrud === false) {
+      throw new Error('Crud instance missing or not an instance of ContentCrud');
+    }
+  }
+
+  get(id: string, options: object) {
+    return this.ctx.crudInstance.get(this.ctx.storageContext, id, options);
+  }
+
+  bulkGet(ids: string[], options: object) {
+    return this.ctx.crudInstance.bulkGet(this.ctx.storageContext, ids, options);
+  }
+
+  create(data: object, options?: object) {
+    return this.ctx.crudInstance.create(this.ctx.storageContext, data, options);
+  }
+
+  update(id: string, data: object, options?: object) {
+    return this.ctx.crudInstance.update(this.ctx.storageContext, id, data, options);
+  }
+
+  delete(id: string, options?: object) {
+    return this.ctx.crudInstance.delete(this.ctx.storageContext, id, options);
+  }
+
+  search(query: object, options?: object) {
+    return this.ctx.crudInstance.search(this.ctx.storageContext, query, options);
+  }
+}

src/plugins/content_management/server/content_client/content_client_factory.ts

@@ -0,0 +1,102 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+import type { RequestHandlerContext } from '@kbn/core-http-request-handler-context-server';
+import type { KibanaRequest } from '@kbn/core-http-server';
+import { Version } from '@kbn/object-versioning';
+
+import type { MSearchIn, MSearchOut } from '../../common';
+import type { ContentRegistry } from '../core';
+import { MSearchService } from '../core/msearch';
+import { getServiceObjectTransformFactory, getStorageContext } from '../utils';
+import { ContentClient } from './content_client';
+
+export const getContentClientFactory =
+  ({ contentRegistry }: { contentRegistry: ContentRegistry }) =>
+  (contentTypeId: string) => {
+    const getForRequest = <T = unknown>({
+      requestHandlerContext,
+      version,
+    }: {
+      requestHandlerContext: RequestHandlerContext;
+      request: KibanaRequest;
+      version?: Version;
+    }) => {
+      const contentDefinition = contentRegistry.getDefinition(contentTypeId);
+
+      const storageContext = getStorageContext({
+        contentTypeId,
+        version: version ?? contentDefinition.version.latest,
+        ctx: {
+          contentRegistry,
+          requestHandlerContext,
+          getTransformsFactory: getServiceObjectTransformFactory,
+        },
+      });
+
+      const crudInstance = contentRegistry.getCrud<T>(contentTypeId);
+
+      return ContentClient.create<T>(contentTypeId, {
+        storageContext,
+        crudInstance,
+      });
+    };
+
+    return {
+      /**
+       * Client getter to interact with the registered content type.
+       */
+      getForRequest,
+    };
+  };
+
+export const getMSearchClientFactory =
+  ({
+    contentRegistry,
+    mSearchService,
+  }: {
+    contentRegistry: ContentRegistry;
+    mSearchService: MSearchService;
+  }) =>
+  ({
+    requestHandlerContext,
+  }: {
+    requestHandlerContext: RequestHandlerContext;
+    request: KibanaRequest;
+  }) => {
+    const msearch = async ({ contentTypes, query }: MSearchIn): Promise<MSearchOut> => {
+      const contentTypesWithStorageContext = contentTypes.map(({ contentTypeId, version }) => {
+        const contentDefinition = contentRegistry.getDefinition(contentTypeId);
+
+        const storageContext = getStorageContext({
+          contentTypeId,
+          version: version ?? contentDefinition.version.latest,
+          ctx: {
+            contentRegistry,
+            requestHandlerContext,
+            getTransformsFactory: getServiceObjectTransformFactory,
+          },
+        });
+
+        return {
+          contentTypeId,
+          ctx: storageContext,
+        };
+      });
+
+      const result = await mSearchService.search(contentTypesWithStorageContext, query);
+
+      return {
+        contentTypes,
+        result,
+      };
+    };
+
+    return {
+      msearch,
+    };
+  };

src/plugins/content_management/server/content_client/index.ts

@@ -6,13 +6,6 @@
  * Side Public License, v 1.
  */
 
-import { Type, ValidationError } from '@kbn/config-schema';
+export { getContentClientFactory, getMSearchClientFactory } from './content_client_factory';
 
-export const validate = (input: unknown, schema: Type<any>): ValidationError | null => {
-  try {
-    schema.validate(input);
-    return null;
-  } catch (e: any) {
-    return e as ValidationError;
-  }
-};
+export type { IContentClient } from './types';

src/plugins/content_management/server/content_client/types.ts

@@ -0,0 +1,53 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { ContentCrud } from '../core/crud';
+
+type CrudGetParameters<T = unknown> = Parameters<ContentCrud<T>['get']>;
+export type GetParameters<T = unknown> = [CrudGetParameters<T>[1], CrudGetParameters<T>[2]?];
+
+type CrudBulkGetParameters<T = unknown> = Parameters<ContentCrud<T>['bulkGet']>;
+export type BulkGetParameters<T = unknown> = [
+  CrudBulkGetParameters<T>[1],
+  CrudBulkGetParameters<T>[2]?
+];
+
+type CrudCreateParameters<T = unknown> = Parameters<ContentCrud<T>['create']>;
+export type CreateParameters<T = unknown> = [
+  CrudCreateParameters<T>[1],
+  CrudCreateParameters<T>[2]?
+];
+
+type CrudUpdateParameters<T = unknown> = Parameters<ContentCrud<T>['update']>;
+export type UpdateParameters<T = unknown> = [
+  CrudUpdateParameters<T>[1],
+  CrudUpdateParameters<T>[2],
+  CrudUpdateParameters<T>[3]?
+];
+
+type CrudDeleteParameters<T = unknown> = Parameters<ContentCrud<T>['delete']>;
+export type DeleteParameters<T = unknown> = [
+  CrudDeleteParameters<T>[1],
+  CrudDeleteParameters<T>[2]?
+];
+
+type CrudSearchParameters<T = unknown> = Parameters<ContentCrud<T>['search']>;
+export type SearchParameters<T = unknown> = [
+  CrudSearchParameters<T>[1],
+  CrudSearchParameters<T>[2]?
+];
+
+export interface IContentClient<T = unknown> {
+  contentTypeId: string;
+  get(...params: GetParameters): ReturnType<ContentCrud<T>['get']>;
+  bulkGet(...params: BulkGetParameters): ReturnType<ContentCrud<T>['bulkGet']>;
+  create(...params: CreateParameters): ReturnType<ContentCrud<T>['create']>;
+  update(...params: UpdateParameters): ReturnType<ContentCrud<T>['update']>;
+  delete(...params: DeleteParameters): ReturnType<ContentCrud<T>['delete']>;
+  search(...params: SearchParameters): ReturnType<ContentCrud<T>['search']>;
+}

src/plugins/content_management/server/core/core.test.ts

@@ -7,8 +7,12 @@
  */
 
 import { loggingSystemMock } from '@kbn/core/server/mocks';
+
+import { until } from '../event_stream/tests/util';
+import { setupEventStreamService } from '../event_stream/tests/setup_event_stream_service';
+import { ContentClient } from '../content_client/content_client';
 import { Core } from './core';
-import { createMemoryStorage } from './mocks';
+import { createMemoryStorage, createMockedStorage } from './mocks';
 import { ContentRegistry } from './registry';
 import type { ContentCrud } from './crud';
 import type {
@@ -31,19 +35,37 @@ import type {
   SearchItemSuccess,
   SearchItemError,
 } from './event_types';
-import { ContentTypeDefinition, StorageContext } from './types';
-import { until } from '../event_stream/tests/util';
-import { setupEventStreamService } from '../event_stream/tests/setup_event_stream_service';
+import { ContentStorage, ContentTypeDefinition, StorageContext } from './types';
+
+const spyMsearch = jest.fn();
+const getmSearchSpy = () => spyMsearch;
+
+jest.mock('./msearch', () => {
+  const original = jest.requireActual('./msearch');
+  class MSearchService {
+    search(...args: any[]) {
+      getmSearchSpy()(...args);
+    }
+  }
+  return {
+    ...original,
+    MSearchService,
+  };
+});
 
 const logger = loggingSystemMock.createLogger();
 
 const FOO_CONTENT_ID = 'foo';
 
-const setup = ({ registerFooType = false }: { registerFooType?: boolean } = {}) => {
+const setup = ({
+  registerFooType = false,
+  storage = createMemoryStorage(),
+  latestVersion = 2,
+}: { registerFooType?: boolean; storage?: ContentStorage; latestVersion?: number } = {}) => {
   const ctx: StorageContext = {
     requestHandlerContext: {} as any,
     version: {
-      latest: 1,
+      latest: latestVersion,
       request: 1,
     },
     utils: {
@@ -60,9 +82,9 @@ const setup = ({ registerFooType = false }: { registerFooType?: boolean } = {})
 
   const contentDefinition: ContentTypeDefinition = {
     id: FOO_CONTENT_ID,
-    storage: createMemoryStorage(),
+    storage,
     version: {
-      latest: 2,
+      latest: latestVersion,
     },
   };
   const cleanUp = () => {
@@ -94,7 +116,12 @@ describe('Content Core', () => {
       const { coreSetup, cleanUp } = setup();
 
       expect(coreSetup.contentRegistry).toBeInstanceOf(ContentRegistry);
-      expect(Object.keys(coreSetup.api).sort()).toEqual(['crud', 'eventBus', 'register']);
+      expect(Object.keys(coreSetup.api).sort()).toEqual([
+        'contentClient',
+        'crud',
+        'eventBus',
+        'register',
+      ]);
 
       cleanUp();
     });
@@ -158,6 +185,52 @@ describe('Content Core', () => {
 
           cleanUp();
         });
+
+        test('should return a contentClient when registering', async () => {
+          const storage = createMockedStorage();
+          const latestVersion = 11;
+
+          const { coreSetup, cleanUp, contentDefinition } = setup({ latestVersion, storage });
+
+          const { contentClient } = coreSetup.api.register(contentDefinition);
+
+          {
+            const client = contentClient.getForRequest({
+              requestHandlerContext: {} as any,
+              request: {} as any,
+              version: 2,
+            });
+
+            await client.get('1234');
+            expect(storage.get).toHaveBeenCalledTimes(1);
+
+            const [storageContext] = storage.get.mock.calls[0];
+            expect(storageContext.version).toEqual({
+              latest: latestVersion,
+              request: 2, // version passed in the request should be used
+            });
+          }
+
+          storage.get.mockReset();
+
+          {
+            // If no request version is passed, the latest version should be used
+            const client = contentClient.getForRequest({
+              requestHandlerContext: {} as any,
+              request: {} as any,
+            });
+
+            await client.get('1234');
+
+            const [storageContext] = storage.get.mock.calls[0];
+            expect(storageContext.version).toEqual({
+              latest: latestVersion,
+              request: latestVersion, // latest version should be used
+            });
+          }
+
+          cleanUp();
+        });
       });
 
       describe('crud()', () => {
@@ -849,38 +922,239 @@ describe('Content Core', () => {
         });
       });
 
-      describe('eventStream', () => {
-        test('stores "delete" events', async () => {
-          const { fooContentCrud, ctx, eventStream } = setup({ registerFooType: true });
+      describe('contentClient', () => {
+        describe('single content type', () => {
+          test('should return a ClientContent instance for a specific request', () => {
+            const { coreSetup, cleanUp } = setup({ registerFooType: true });
 
-          await fooContentCrud!.create(ctx, { title: 'Hello' }, { id: '1234' });
-          await fooContentCrud!.delete(ctx, '1234');
+            const {
+              api: { contentClient },
+            } = coreSetup;
 
-          const findEvent = async () => {
-            const tail = await eventStream.tail();
+            const client = contentClient
+              .getForRequest({
+                requestHandlerContext: {} as any,
+                request: {} as any,
+              })
+              .for(FOO_CONTENT_ID);
 
-            for (const event of tail) {
-              if (
-                event.predicate[0] === 'delete' &&
-                event.object &&
-                event.object[0] === 'foo' &&
-                event.object[1] === '1234'
-              ) {
-                return event;
-              }
-            }
+            expect(client).toBeInstanceOf(ContentClient);
 
-            return null;
-          };
-
-          await until(async () => !!(await findEvent()), 100);
-
-          const event = await findEvent();
-
-          expect(event).toMatchObject({
-            predicate: ['delete'],
-            object: ['foo', '1234'],
+            cleanUp();
           });
+
+          test('should automatically set the content version to the latest version if not provided', async () => {
+            const storage = createMockedStorage();
+            const latestVersion = 7;
+
+            const { coreSetup, cleanUp } = setup({
+              registerFooType: true,
+              latestVersion,
+              storage,
+            });
+
+            const requestHandlerContext = {} as any;
+            const client = coreSetup.api.contentClient
+              .getForRequest({
+                requestHandlerContext,
+                request: {} as any,
+              })
+              .for(FOO_CONTENT_ID);
+
+            const options = { foo: 'bar' };
+            await client.get('1234', options);
+
+            const storageContext = {
+              requestHandlerContext,
+              utils: { getTransforms: expect.any(Function) },
+              version: {
+                latest: latestVersion,
+                request: latestVersion, // Request version should be set to the latest version
+              },
+            };
+
+            expect(storage.get).toHaveBeenCalledWith(storageContext, '1234', options);
+
+            cleanUp();
+          });
+
+          test('should pass the provided content version', async () => {
+            const storage = createMockedStorage();
+            const latestVersion = 7;
+            const requestVersion = 2;
+
+            const { coreSetup, cleanUp } = setup({
+              registerFooType: true,
+              latestVersion,
+              storage,
+            });
+
+            const requestHandlerContext = {} as any;
+
+            const client = coreSetup.api.contentClient
+              .getForRequest({
+                requestHandlerContext,
+                request: {} as any,
+              })
+              .for(FOO_CONTENT_ID, requestVersion);
+
+            await client.get('1234');
+
+            const storageContext = {
+              requestHandlerContext,
+              utils: { getTransforms: expect.any(Function) },
+              version: {
+                latest: latestVersion,
+                request: requestVersion, // The requested version should be used
+              },
+            };
+
+            expect(storage.get).toHaveBeenCalledWith(storageContext, '1234', undefined);
+
+            cleanUp();
+          });
+
+          test('should throw if the contentTypeId is not registered', () => {
+            const { coreSetup, cleanUp } = setup();
+
+            const {
+              api: { contentClient },
+            } = coreSetup;
+
+            expect(() => {
+              contentClient
+                .getForRequest({
+                  requestHandlerContext: {} as any,
+                  request: {} as any,
+                })
+                .for(FOO_CONTENT_ID);
+            }).toThrowError('Content [foo] is not registered.');
+
+            cleanUp();
+          });
+        });
+
+        describe('multiple content types', () => {
+          const storage = createMockedStorage();
+
+          beforeEach(() => {
+            spyMsearch.mockReset();
+          });
+
+          test('should automatically set the content version to the latest version if not provided', async () => {
+            const { coreSetup, cleanUp } = setup();
+
+            coreSetup.api.register({
+              id: 'foo',
+              storage,
+              version: {
+                latest: 9, // Needs to be automatically passed to the mSearch service
+              },
+            });
+
+            coreSetup.api.register({
+              id: 'bar',
+              storage,
+              version: {
+                latest: 11, // Needs to be automatically passed to the mSearch service
+              },
+            });
+
+            const client = coreSetup.api.contentClient.getForRequest({
+              requestHandlerContext: {} as any,
+              request: {} as any,
+            });
+
+            await client.msearch({
+              // We don't pass the version here
+              contentTypes: [{ contentTypeId: 'foo' }, { contentTypeId: 'bar' }],
+              query: { text: 'Hello' },
+            });
+
+            const [contentTypes] = spyMsearch.mock.calls[0];
+            expect(contentTypes[0].contentTypeId).toBe('foo');
+            expect(contentTypes[0].ctx.version).toEqual({ latest: 9, request: 9 });
+
+            expect(contentTypes[1].contentTypeId).toBe('bar');
+            expect(contentTypes[1].ctx.version).toEqual({ latest: 11, request: 11 });
+
+            cleanUp();
+          });
+
+          test('should use the request version if provided', async () => {
+            const { coreSetup, cleanUp } = setup();
+
+            coreSetup.api.register({
+              id: 'foo',
+              storage,
+              version: {
+                latest: 9, // Needs to be automatically passed to the mSearch service
+              },
+            });
+
+            coreSetup.api.register({
+              id: 'bar',
+              storage,
+              version: {
+                latest: 11, // Needs to be automatically passed to the mSearch service
+              },
+            });
+
+            const client = coreSetup.api.contentClient.getForRequest({
+              requestHandlerContext: {} as any,
+              request: {} as any,
+            });
+
+            await client.msearch({
+              // We don't pass the version here
+              contentTypes: [
+                { contentTypeId: 'foo', version: 2 },
+                { contentTypeId: 'bar', version: 3 },
+              ],
+              query: { text: 'Hello' },
+            });
+
+            const [contentTypes] = spyMsearch.mock.calls[0];
+            expect(contentTypes[0].ctx.version).toEqual({ latest: 9, request: 2 });
+            expect(contentTypes[1].ctx.version).toEqual({ latest: 11, request: 3 });
+
+            cleanUp();
+          });
+        });
+      });
+    });
+
+    describe('eventStream', () => {
+      test('stores "delete" events', async () => {
+        const { fooContentCrud, ctx, eventStream } = setup({ registerFooType: true });
+
+        await fooContentCrud!.create(ctx, { title: 'Hello' }, { id: '1234' });
+        await fooContentCrud!.delete(ctx, '1234');
+
+        const findEvent = async () => {
+          const tail = await eventStream.tail();
+
+          for (const event of tail) {
+            if (
+              event.predicate[0] === 'delete' &&
+              event.object &&
+              event.object[0] === 'foo' &&
+              event.object[1] === '1234'
+            ) {
+              return event;
+            }
+          }
+
+          return null;
+        };
+
+        await until(async () => !!(await findEvent()), 100);
+
+        const event = await findEvent();
+
+        expect(event).toMatchObject({
+          predicate: ['delete'],
+          object: ['foo', '1234'],
         });
       });
     });

src/plugins/content_management/server/core/core.ts

@@ -6,11 +6,27 @@
  * Side Public License, v 1.
  */
 
-import { Logger } from '@kbn/core/server';
+import type { Logger, KibanaRequest } from '@kbn/core/server';
+import type { RequestHandlerContext } from '@kbn/core-http-request-handler-context-server';
+import type { Version } from '@kbn/object-versioning';
+import { LISTING_LIMIT_SETTING, PER_PAGE_SETTING } from '@kbn/saved-objects-settings';
+
+import type { MSearchIn, MSearchOut } from '../../common';
+import {
+  getContentClientFactory,
+  getMSearchClientFactory,
+  IContentClient,
+} from '../content_client';
 import { EventStreamService } from '../event_stream';
 import { ContentCrud } from './crud';
 import { EventBus } from './event_bus';
 import { ContentRegistry } from './registry';
+import { MSearchService } from './msearch';
+
+export interface GetContentClientForRequestDependencies {
+  requestHandlerContext: RequestHandlerContext;
+  request: KibanaRequest;
+}
 
 export interface CoreApi {
   /**
@@ -24,6 +40,14 @@ export interface CoreApi {
   crud: <T = unknown>(contentType: string) => ContentCrud<T>;
   /** Content management event bus */
   eventBus: EventBus;
+  /** Client getters to interact with registered content types. */
+  contentClient: {
+    /** Client getter to interact with registered content types for the current HTTP request. */
+    getForRequest(deps: GetContentClientForRequestDependencies): {
+      for: <T = unknown>(contentTypeId: string, version?: Version) => IContentClient<T>;
+      msearch(args: MSearchIn): Promise<MSearchOut>;
+    };
+  };
 }
 
 export interface CoreInitializerContext {
@@ -52,13 +76,18 @@ export class Core {
   setup(): CoreSetup {
     this.setupEventStream();
 
+    const coreApi: CoreApi = {
+      register: this.contentRegistry.register.bind(this.contentRegistry),
+      crud: this.contentRegistry.getCrud.bind(this.contentRegistry),
+      eventBus: this.eventBus,
+      contentClient: {
+        getForRequest: this.getContentClientForRequest.bind(this),
+      },
+    };
+
     return {
       contentRegistry: this.contentRegistry,
-      api: {
-        register: this.contentRegistry.register.bind(this.contentRegistry),
-        crud: this.contentRegistry.getCrud.bind(this.contentRegistry),
-        eventBus: this.eventBus,
-      },
+      api: coreApi,
     };
   }
 
@@ -79,4 +108,50 @@ export class Core {
       });
     }
   }
+
+  private getContentClientForRequest({
+    requestHandlerContext,
+    request,
+  }: {
+    request: KibanaRequest;
+    requestHandlerContext: RequestHandlerContext;
+  }) {
+    /** Handler to return a ContentClient for a specific content type Id and request version */
+    const forFn = <T = unknown>(contentTypeId: string, version?: Version) => {
+      const contentDefinition = this.contentRegistry.getDefinition(contentTypeId);
+      const clientFactory = getContentClientFactory({
+        contentRegistry: this.contentRegistry,
+      });
+      const contentClient = clientFactory(contentTypeId);
+
+      return contentClient.getForRequest<T>({
+        requestHandlerContext,
+        request,
+        version: version ?? contentDefinition.version.latest,
+      });
+    };
+
+    const mSearchService = new MSearchService({
+      getSavedObjectsClient: async () => (await requestHandlerContext.core).savedObjects.client,
+      contentRegistry: this.contentRegistry,
+      getConfig: {
+        listingLimit: async () =>
+          (await requestHandlerContext.core).uiSettings.client.get(LISTING_LIMIT_SETTING),
+        perPage: async () =>
+          (await requestHandlerContext.core).uiSettings.client.get(PER_PAGE_SETTING),
+      },
+    });
+
+    const msearchClientFactory = getMSearchClientFactory({
+      contentRegistry: this.contentRegistry,
+      mSearchService,
+    });
+
+    const msearchClient = msearchClientFactory({ requestHandlerContext, request });
+
+    return {
+      for: forFn,
+      msearch: msearchClient.msearch,
+    };
+  }
 }

src/plugins/content_management/server/core/crud.ts

@@ -79,17 +79,17 @@ export class ContentCrud<T = unknown> {
     });
 
     try {
-      const item = await this.storage.get(ctx, contentId, options);
+      const result = await this.storage.get(ctx, contentId, options);
 
       this.eventBus.emit({
         type: 'getItemSuccess',
         contentId,
         contentTypeId: this.contentTypeId,
-        data: item,
+        data: result,
         options,
       });
 
-      return { contentTypeId: this.contentTypeId, result: item };
+      return { contentTypeId: this.contentTypeId, result };
     } catch (e) {
       this.eventBus.emit({
         type: 'getItemError',

src/plugins/content_management/server/core/mocks/in_memory_storage.ts

@@ -64,7 +64,11 @@ class InMemoryStorage implements ContentStorage<any> {
   async create(
     ctx: StorageContext,
     data: Omit<FooContent, 'id'>,
-    { id: _id, errorToThrow }: { id?: string; errorToThrow?: string } = {}
+    {
+      id: _id,
+      forwardInResponse,
+      errorToThrow,
+    }: { id?: string; errorToThrow?: string; forwardInResponse?: object } = {}
   ) {
     // This allows us to test that proper error events are thrown when the storage layer op fails
     if (errorToThrow) {
@@ -81,6 +85,16 @@ class InMemoryStorage implements ContentStorage<any> {
 
     this.db.set(id, content);
 
+    if (forwardInResponse) {
+      // We add this so we can test that options are passed down to the storage layer
+      return {
+        item: {
+          ...content,
+          options: forwardInResponse,
+        },
+      };
+    }
+
     return {
       item: content,
     };
@@ -159,7 +173,7 @@ class InMemoryStorage implements ContentStorage<any> {
   async search(
     ctx: StorageContext,
     query: { text: string },
-    { errorToThrow }: { errorToThrow?: string } = {}
+    { errorToThrow, forwardInResponse }: { errorToThrow?: string; forwardInResponse?: object } = {}
   ) {
     // This allows us to test that proper error events are thrown when the storage layer op fails
     if (errorToThrow) {
@@ -181,7 +195,7 @@ class InMemoryStorage implements ContentStorage<any> {
       return title.match(rgx);
     });
     return {
-      hits,
+      hits: forwardInResponse ? hits.map((hit) => ({ ...hit, options: forwardInResponse })) : hits,
       pagination: {
         total: hits.length,
         cursor: '',
@@ -190,8 +204,8 @@ class InMemoryStorage implements ContentStorage<any> {
   }
 }
 
-export const createMemoryStorage = () => {
-  return new InMemoryStorage();
+export const createMemoryStorage = (): ContentStorage<FooContent> => {
+  return new InMemoryStorage() as ContentStorage<FooContent>;
 };
 
 export const createMockedStorage = (): jest.Mocked<ContentStorage> => ({

src/plugins/content_management/server/core/registry.ts

@@ -7,6 +7,8 @@
  */
 
 import { validateVersion } from '@kbn/object-versioning/lib/utils';
+
+import { getContentClientFactory } from '../content_client';
 import { ContentType } from './content_type';
 import { EventBus } from './event_bus';
 import type { ContentStorage, ContentTypeDefinition, MSearchConfig } from './types';
@@ -45,6 +47,15 @@ export class ContentRegistry {
     );
 
     this.types.set(contentType.id, contentType);
+
+    const contentClient = getContentClientFactory({ contentRegistry: this })(contentType.id);
+
+    return {
+      /**
+       * Client getters to interact with the registered content type.
+       */
+      contentClient,
+    };
   }
 
   getContentType(id: string): ContentType {

src/plugins/content_management/server/core/types.ts

@@ -31,12 +31,26 @@ export type StorageContextGetTransformFn = (
 
 /** Context that is sent to all storage instance methods */
 export interface StorageContext {
+  /** The Core HTTP request handler context */
   requestHandlerContext: RequestHandlerContext;
   version: {
+    /**
+     * The content type version for the request. It usually is the latest version although in some
+     * cases the client (browser) might still be on an older version and make requests with that version.
+     */
     request: Version;
+    /**
+     * The latest version of the content type. This is the version that the content type is currently on
+     * after updating the Kibana server.
+     */
     latest: Version;
   };
   utils: {
+    /**
+     * Get the transforms handlers for the content type.
+     * The transforms are used to transform the content object to the latest schema (up) and back
+     * to a previous schema (down).
+     */
     getTransforms: StorageContextGetTransformFn;
   };
 }

src/plugins/content_management/server/plugin.test.ts

@@ -138,8 +138,8 @@ describe('ContentManagementPlugin', () => {
         // Each procedure has been called with the context and input
         const context = {
           requestHandlerContext: mockedRequestHandlerContext,
+          request: expect.any(Object),
           contentRegistry: 'mockedContentRegistry',
-          getTransformsFactory: expect.any(Function),
           mSearchService: expect.any(MSearchService),
         };
         expect(mockGet).toHaveBeenCalledWith(context, input);

src/plugins/content_management/server/rpc/procedures/bulk_get.test.ts

@@ -9,14 +9,14 @@
 import { omit } from 'lodash';
 
 import { schema } from '@kbn/config-schema';
-import type { ContentManagementServiceDefinitionVersioned, Version } from '@kbn/object-versioning';
-import { validate } from '../../utils';
+import type { ContentManagementServiceDefinitionVersioned } from '@kbn/object-versioning';
+import { validate, disableTransformsCache } from '../../utils';
 import { ContentRegistry } from '../../core/registry';
 import { createMockedStorage } from '../../core/mocks';
 import { EventBus } from '../../core/event_bus';
-import { getServiceObjectTransformFactory } from '../services_transforms_factory';
 import { bulkGet } from './bulk_get';
 
+disableTransformsCache();
 const storageContextGetTransforms = jest.fn();
 const spy = () => storageContextGetTransforms;
 
@@ -191,8 +191,6 @@ describe('RPC -> bulkGet()', () => {
       const ctx: any = {
         contentRegistry,
         requestHandlerContext,
-        getTransformsFactory: (contentTypeId: string, version: Version) =>
-          getServiceObjectTransformFactory(contentTypeId, version, { cacheEnabled: false }),
       };
 
       return { ctx, storage };

src/plugins/content_management/server/rpc/procedures/bulk_get.ts

@@ -11,18 +11,19 @@ import type { BulkGetIn } from '../../../common';
 import type { ProcedureDefinition } from '../rpc_service';
 import type { Context } from '../types';
 import { BulkGetResponse } from '../../core/crud';
-import { getStorageContext } from './utils';
+import { getContentClientFactory } from '../../content_client';
 
 export const bulkGet: ProcedureDefinition<Context, BulkGetIn<string>, BulkGetResponse> = {
   schemas: rpcSchemas.bulkGet,
   fn: async (ctx, { contentTypeId, version, ids, options }) => {
-    const storageContext = getStorageContext({
-      contentTypeId,
+    const clientFactory = getContentClientFactory({
+      contentRegistry: ctx.contentRegistry,
+    });
+    const client = clientFactory(contentTypeId).getForRequest({
+      ...ctx,
       version,
-      ctx,
     });
 
-    const crudInstance = ctx.contentRegistry.getCrud(contentTypeId);
-    return crudInstance.bulkGet(storageContext, ids, options);
+    return client.bulkGet(ids, options);
   },
 };

src/plugins/content_management/server/rpc/procedures/create.test.ts

@@ -9,14 +9,14 @@
 import { omit } from 'lodash';
 
 import { schema } from '@kbn/config-schema';
-import type { ContentManagementServiceDefinitionVersioned, Version } from '@kbn/object-versioning';
-import { validate } from '../../utils';
+import type { ContentManagementServiceDefinitionVersioned } from '@kbn/object-versioning';
+import { validate, disableTransformsCache } from '../../utils';
 import { ContentRegistry } from '../../core/registry';
 import { createMockedStorage } from '../../core/mocks';
 import { EventBus } from '../../core/event_bus';
-import { getServiceObjectTransformFactory } from '../services_transforms_factory';
 import { create } from './create';
 
+disableTransformsCache();
 const storageContextGetTransforms = jest.fn();
 const spy = () => storageContextGetTransforms;
 
@@ -164,8 +164,6 @@ describe('RPC -> create()', () => {
       const ctx: any = {
         contentRegistry,
         requestHandlerContext,
-        getTransformsFactory: (contentTypeId: string, version: Version) =>
-          getServiceObjectTransformFactory(contentTypeId, version, { cacheEnabled: false }),
       };
 
       return { ctx, storage };

src/plugins/content_management/server/rpc/procedures/create.ts

@@ -8,21 +8,21 @@
 
 import { rpcSchemas } from '../../../common/schemas';
 import type { CreateIn } from '../../../common';
+import { getContentClientFactory } from '../../content_client';
 import type { ProcedureDefinition } from '../rpc_service';
 import type { Context } from '../types';
-import { getStorageContext } from './utils';
 
 export const create: ProcedureDefinition<Context, CreateIn<string>> = {
   schemas: rpcSchemas.create,
   fn: async (ctx, { contentTypeId, version, data, options }) => {
-    const storageContext = getStorageContext({
-      contentTypeId,
+    const clientFactory = getContentClientFactory({
+      contentRegistry: ctx.contentRegistry,
+    });
+    const client = clientFactory(contentTypeId).getForRequest({
+      ...ctx,
       version,
-      ctx,
     });
 
-    const crudInstance = ctx.contentRegistry.getCrud(contentTypeId);
-
-    return crudInstance.create(storageContext, data, options);
+    return client.create(data, options);
   },
 };

src/plugins/content_management/server/rpc/procedures/delete.test.ts

@@ -8,15 +8,15 @@
 
 import { omit } from 'lodash';
 
-import type { ContentManagementServiceDefinitionVersioned, Version } from '@kbn/object-versioning';
+import type { ContentManagementServiceDefinitionVersioned } from '@kbn/object-versioning';
 import { schema } from '@kbn/config-schema';
-import { validate } from '../../utils';
+import { validate, disableTransformsCache } from '../../utils';
 import { ContentRegistry } from '../../core/registry';
 import { createMockedStorage } from '../../core/mocks';
 import { EventBus } from '../../core/event_bus';
-import { getServiceObjectTransformFactory } from '../services_transforms_factory';
 import { deleteProc } from './delete';
 
+disableTransformsCache();
 const storageContextGetTransforms = jest.fn();
 const spy = () => storageContextGetTransforms;
 
@@ -150,8 +150,6 @@ describe('RPC -> delete()', () => {
       const ctx: any = {
         contentRegistry,
         requestHandlerContext,
-        getTransformsFactory: (contentTypeId: string, version: Version) =>
-          getServiceObjectTransformFactory(contentTypeId, version, { cacheEnabled: false }),
       };
 
       return { ctx, storage };

src/plugins/content_management/server/rpc/procedures/delete.ts

@@ -7,20 +7,21 @@
  */
 import { rpcSchemas } from '../../../common/schemas';
 import type { DeleteIn } from '../../../common';
+import { getContentClientFactory } from '../../content_client';
 import type { ProcedureDefinition } from '../rpc_service';
 import type { Context } from '../types';
-import { getStorageContext } from './utils';
 
 export const deleteProc: ProcedureDefinition<Context, DeleteIn<string>> = {
   schemas: rpcSchemas.delete,
   fn: async (ctx, { contentTypeId, id, version, options }) => {
-    const storageContext = getStorageContext({
-      contentTypeId,
+    const clientFactory = getContentClientFactory({
+      contentRegistry: ctx.contentRegistry,
+    });
+    const client = clientFactory(contentTypeId).getForRequest({
+      ...ctx,
       version,
-      ctx,
     });
 
-    const crudInstance = ctx.contentRegistry.getCrud(contentTypeId);
-    return crudInstance.delete(storageContext, id, options);
+    return client.delete(id, options);
   },
 };

src/plugins/content_management/server/rpc/procedures/get.test.ts

@@ -9,14 +9,14 @@
 import { omit } from 'lodash';
 
 import { schema } from '@kbn/config-schema';
-import type { ContentManagementServiceDefinitionVersioned, Version } from '@kbn/object-versioning';
-import { validate } from '../../utils';
+import type { ContentManagementServiceDefinitionVersioned } from '@kbn/object-versioning';
+import { validate, disableTransformsCache } from '../../utils';
 import { ContentRegistry } from '../../core/registry';
 import { createMockedStorage } from '../../core/mocks';
 import { EventBus } from '../../core/event_bus';
-import { getServiceObjectTransformFactory } from '../services_transforms_factory';
 import { get } from './get';
 
+disableTransformsCache();
 const storageContextGetTransforms = jest.fn();
 const spy = () => storageContextGetTransforms;
 
@@ -157,8 +157,6 @@ describe('RPC -> get()', () => {
       const ctx: any = {
         contentRegistry,
         requestHandlerContext,
-        getTransformsFactory: (contentTypeId: string, version: Version) =>
-          getServiceObjectTransformFactory(contentTypeId, version, { cacheEnabled: false }),
       };
 
       return { ctx, storage };

src/plugins/content_management/server/rpc/procedures/get.ts

@@ -8,20 +8,21 @@
 
 import { rpcSchemas } from '../../../common/schemas';
 import type { GetIn } from '../../../common';
+import { getContentClientFactory } from '../../content_client';
 import type { ProcedureDefinition } from '../rpc_service';
 import type { Context } from '../types';
-import { getStorageContext } from './utils';
 
 export const get: ProcedureDefinition<Context, GetIn<string>> = {
   schemas: rpcSchemas.get,
   fn: async (ctx, { contentTypeId, id, version, options }) => {
-    const storageContext = getStorageContext({
-      contentTypeId,
+    const clientFactory = getContentClientFactory({
+      contentRegistry: ctx.contentRegistry,
+    });
+    const client = clientFactory(contentTypeId).getForRequest({
+      ...ctx,
       version,
-      ctx,
     });
 
-    const crudInstance = ctx.contentRegistry.getCrud(contentTypeId);
-    return crudInstance.get(storageContext, id, options);
+    return client.get(id, options);
   },
 };

src/plugins/content_management/server/rpc/procedures/msearch.test.ts

@@ -6,18 +6,17 @@
  * Side Public License, v 1.
  */
 
-import type { Version } from '@kbn/object-versioning';
 import { savedObjectsClientMock } from '@kbn/core-saved-objects-api-server-mocks';
 
 import { MSearchIn, MSearchQuery } from '../../../common';
-import { validate } from '../../utils';
+import { validate, disableTransformsCache } from '../../utils';
 import { ContentRegistry } from '../../core/registry';
 import { createMockedStorage } from '../../core/mocks';
 import { EventBus } from '../../core/event_bus';
 import { MSearchService } from '../../core/msearch';
-import { getServiceObjectTransformFactory } from '../services_transforms_factory';
 import { mSearch } from './msearch';
 
+disableTransformsCache();
 const storageContextGetTransforms = jest.fn();
 const spy = () => storageContextGetTransforms;
 
@@ -151,8 +150,6 @@ describe('RPC -> mSearch()', () => {
       const ctx: any = {
         contentRegistry,
         requestHandlerContext,
-        getTransformsFactory: (contentTypeId: string, version: Version) =>
-          getServiceObjectTransformFactory(contentTypeId, version, { cacheEnabled: false }),
         mSearchService,
       };
 

src/plugins/content_management/server/rpc/procedures/msearch.ts

@@ -10,29 +10,17 @@ import { rpcSchemas } from '../../../common/schemas';
 import type { MSearchIn, MSearchOut } from '../../../common';
 import type { ProcedureDefinition } from '../rpc_service';
 import type { Context } from '../types';
-import { getStorageContext } from './utils';
+import { getMSearchClientFactory } from '../../content_client';
 
 export const mSearch: ProcedureDefinition<Context, MSearchIn, MSearchOut> = {
   schemas: rpcSchemas.mSearch,
-  fn: async (ctx, { contentTypes: contentTypes, query }) => {
-    const contentTypesWithStorageContext = contentTypes.map(({ contentTypeId, version }) => {
-      const storageContext = getStorageContext({
-        contentTypeId,
-        version,
-        ctx,
-      });
-
-      return {
-        contentTypeId,
-        ctx: storageContext,
-      };
+  fn: async (ctx, { contentTypes, query }) => {
+    const clientFactory = getMSearchClientFactory({
+      contentRegistry: ctx.contentRegistry,
+      mSearchService: ctx.mSearchService,
     });
+    const mSearchClient = clientFactory(ctx);
 
-    const result = await ctx.mSearchService.search(contentTypesWithStorageContext, query);
-
-    return {
-      contentTypes,
-      result,
-    };
+    return mSearchClient.msearch({ contentTypes, query });
   },
 };

src/plugins/content_management/server/rpc/procedures/search.test.ts

@@ -8,16 +8,16 @@
 
 import { omit } from 'lodash';
 import { schema } from '@kbn/config-schema';
-import type { ContentManagementServiceDefinitionVersioned, Version } from '@kbn/object-versioning';
+import type { ContentManagementServiceDefinitionVersioned } from '@kbn/object-versioning';
 
 import type { SearchQuery } from '../../../common';
-import { validate } from '../../utils';
+import { validate, disableTransformsCache } from '../../utils';
 import { ContentRegistry } from '../../core/registry';
 import { createMockedStorage } from '../../core/mocks';
 import { EventBus } from '../../core/event_bus';
-import { getServiceObjectTransformFactory } from '../services_transforms_factory';
 import { search } from './search';
 
+disableTransformsCache();
 const storageContextGetTransforms = jest.fn();
 const spy = () => storageContextGetTransforms;
 
@@ -177,8 +177,6 @@ describe('RPC -> search()', () => {
       const ctx: any = {
         contentRegistry,
         requestHandlerContext,
-        getTransformsFactory: (contentTypeId: string, version: Version) =>
-          getServiceObjectTransformFactory(contentTypeId, version, { cacheEnabled: false }),
       };
 
       return { ctx, storage };

src/plugins/content_management/server/rpc/procedures/search.ts

@@ -8,20 +8,21 @@
 
 import { rpcSchemas } from '../../../common/schemas';
 import type { SearchIn } from '../../../common';
+import { getContentClientFactory } from '../../content_client';
 import type { ProcedureDefinition } from '../rpc_service';
 import type { Context } from '../types';
-import { getStorageContext } from './utils';
 
 export const search: ProcedureDefinition<Context, SearchIn<string>> = {
   schemas: rpcSchemas.search,
   fn: async (ctx, { contentTypeId, version, query, options }) => {
-    const storageContext = getStorageContext({
-      contentTypeId,
+    const clientFactory = getContentClientFactory({
+      contentRegistry: ctx.contentRegistry,
+    });
+    const client = clientFactory(contentTypeId).getForRequest({
+      ...ctx,
       version,
-      ctx,
     });
 
-    const crudInstance = ctx.contentRegistry.getCrud(contentTypeId);
-    return crudInstance.search(storageContext, query, options);
+    return client.search(query, options);
   },
 };

src/plugins/content_management/server/rpc/procedures/update.test.ts

@@ -8,15 +8,15 @@
 
 import { omit } from 'lodash';
 import { schema } from '@kbn/config-schema';
-import type { ContentManagementServiceDefinitionVersioned, Version } from '@kbn/object-versioning';
+import type { ContentManagementServiceDefinitionVersioned } from '@kbn/object-versioning';
 
-import { validate } from '../../utils';
+import { validate, disableTransformsCache } from '../../utils';
 import { ContentRegistry } from '../../core/registry';
 import { createMockedStorage } from '../../core/mocks';
 import { EventBus } from '../../core/event_bus';
-import { getServiceObjectTransformFactory } from '../services_transforms_factory';
 import { update } from './update';
 
+disableTransformsCache();
 const storageContextGetTransforms = jest.fn();
 const spy = () => storageContextGetTransforms;
 
@@ -171,8 +171,6 @@ describe('RPC -> update()', () => {
       const ctx: any = {
         contentRegistry,
         requestHandlerContext,
-        getTransformsFactory: (contentTypeId: string, version: Version) =>
-          getServiceObjectTransformFactory(contentTypeId, version, { cacheEnabled: false }),
       };
 
       return { ctx, storage };

src/plugins/content_management/server/rpc/procedures/update.ts

@@ -7,19 +7,21 @@
  */
 import { rpcSchemas } from '../../../common/schemas';
 import type { UpdateIn } from '../../../common';
+import { getContentClientFactory } from '../../content_client';
 import type { ProcedureDefinition } from '../rpc_service';
 import type { Context } from '../types';
-import { getStorageContext } from './utils';
 
 export const update: ProcedureDefinition<Context, UpdateIn<string>> = {
   schemas: rpcSchemas.update,
   fn: async (ctx, { contentTypeId, id, version, data, options }) => {
-    const storageContext = getStorageContext({
-      contentTypeId,
-      version,
-      ctx,
+    const clientFactory = getContentClientFactory({
+      contentRegistry: ctx.contentRegistry,
     });
-    const crudInstance = ctx.contentRegistry.getCrud(contentTypeId);
-    return crudInstance.update(storageContext, id, data, options);
+    const client = clientFactory(contentTypeId).getForRequest({
+      ...ctx,
+      version,
+    });
+
+    return client.update(id, data, options);
   },
 };

src/plugins/content_management/server/rpc/routes/routes.ts

@@ -14,7 +14,6 @@ import type { ContentRegistry } from '../../core';
 import { MSearchService } from '../../core/msearch';
 
 import type { RpcService } from '../rpc_service';
-import { getServiceObjectTransformFactory } from '../services_transforms_factory';
 import type { Context as RpcContext } from '../types';
 import { wrapError } from './error_wrapper';
 
@@ -56,7 +55,7 @@ export function initRpcRoutes(
         const context: RpcContext = {
           contentRegistry,
           requestHandlerContext,
-          getTransformsFactory: getServiceObjectTransformFactory,
+          request,
           mSearchService: new MSearchService({
             getSavedObjectsClient: async () =>
               (await requestHandlerContext.core).savedObjects.client,

src/plugins/content_management/server/rpc/types.ts

@@ -6,17 +6,13 @@
  * Side Public License, v 1.
  */
 import type { RequestHandlerContext } from '@kbn/core-http-request-handler-context-server';
-import type { Version } from '@kbn/object-versioning';
-import type { ContentRegistry, StorageContextGetTransformFn } from '../core';
+import type { KibanaRequest } from '@kbn/core-http-server';
+import type { ContentRegistry } from '../core';
 import type { MSearchService } from '../core/msearch';
 
 export interface Context {
   contentRegistry: ContentRegistry;
   requestHandlerContext: RequestHandlerContext;
-  getTransformsFactory: (
-    contentTypeId: string,
-    requestVersion: Version,
-    options?: { cacheEnabled?: boolean }
-  ) => StorageContextGetTransformFn;
+  request: KibanaRequest;
   mSearchService: MSearchService;
 }

src/plugins/content_management/server/types.ts

@@ -6,7 +6,8 @@
  * Side Public License, v 1.
  */
 
-import { CoreApi } from './core';
+import type { Version } from '@kbn/object-versioning';
+import type { CoreApi, StorageContextGetTransformFn } from './core';
 
 // eslint-disable-next-line @typescript-eslint/no-empty-interface
 export interface ContentManagementServerSetupDependencies {}
@@ -19,3 +20,9 @@ export interface ContentManagementServerSetup extends CoreApi {}
 
 // eslint-disable-next-line @typescript-eslint/no-empty-interface
 export interface ContentManagementServerStart {}
+
+export type GetTransformsFactoryFn = (
+  contentTypeId: string,
+  requestVersion: Version,
+  options?: { cacheEnabled?: boolean }
+) => StorageContextGetTransformFn;

src/plugins/content_management/server/utils/index.ts

@@ -0,0 +1,14 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+export { getStorageContext, validate } from './utils';
+
+export {
+  getServiceObjectTransformFactory,
+  disableCache as disableTransformsCache,
+} from './services_transforms_factory';

src/plugins/content_management/server/utils/services_transforms_factory.ts

@@ -15,6 +15,13 @@ import {
 } from '@kbn/object-versioning';
 import type { StorageContextGetTransformFn } from '../core';
 
+let isCacheEnabled = true;
+
+// This is used in tests to disable the cache
+export const disableCache = () => {
+  isCacheEnabled = false;
+};
+
 /**
  * We keep a cache of compiled service definition to avoid unnecessary recompile on every request.
  */
@@ -32,15 +39,11 @@ const compiledCache = new LRUCache<string, { [path: string]: ObjectMigrationDefi
  * @returns A "getContentManagmentServicesTransforms()"
  */
 export const getServiceObjectTransformFactory =
-  (
-    contentTypeId: string,
-    _requestVersion: Version,
-    { cacheEnabled = true }: { cacheEnabled?: boolean } = {}
-  ): StorageContextGetTransformFn =>
+  (contentTypeId: string, _requestVersion: Version): StorageContextGetTransformFn =>
   (definitions: ContentManagementServiceDefinitionVersioned, requestVersionOverride?: Version) => {
     const requestVersion = requestVersionOverride ?? _requestVersion;
 
-    if (cacheEnabled) {
+    if (isCacheEnabled) {
       const compiledFromCache = compiledCache.get(contentTypeId);
 
       if (compiledFromCache) {
@@ -54,7 +57,7 @@ export const getServiceObjectTransformFactory =
 
     const compiled = compileServiceDefinitions(definitions);
 
-    if (cacheEnabled) {
+    if (isCacheEnabled) {
       compiledCache.set(contentTypeId, compiled);
     }
 

src/plugins/content_management/server/utils/utils.ts

@@ -6,10 +6,22 @@
  * Side Public License, v 1.
  */
 
+import { Type, ValidationError } from '@kbn/config-schema';
+import type { RequestHandlerContext } from '@kbn/core-http-request-handler-context-server';
 import { validateVersion } from '@kbn/object-versioning/lib/utils';
 import type { Version } from '@kbn/object-versioning';
-import type { StorageContext } from '../../core';
-import type { Context as RpcContext } from '../types';
+
+import type { ContentRegistry, StorageContext } from '../core';
+import type { GetTransformsFactoryFn } from '../types';
+
+export const validate = (input: unknown, schema: Type<any>): ValidationError | null => {
+  try {
+    schema.validate(input);
+    return null;
+  } catch (e: any) {
+    return e as ValidationError;
+  }
+};
 
 const validateRequestVersion = (
   requestVersion: Version | undefined,
@@ -40,7 +52,11 @@ export const getStorageContext = ({
 }: {
   contentTypeId: string;
   version?: number;
-  ctx: RpcContext;
+  ctx: {
+    contentRegistry: ContentRegistry;
+    requestHandlerContext: RequestHandlerContext;
+    getTransformsFactory: GetTransformsFactoryFn;
+  };
 }): StorageContext => {
   const contentDefinition = contentRegistry.getDefinition(contentTypeId);
   const version = validateRequestVersion(_version, contentDefinition.version.latest);