🌊 Add task service and client to manage background tasks

x-pack/platform/plugins/shared/streams/server/lib/tasks/storage.ts

@@ -0,0 +1,28 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { types, type IndexStorageSettings, type IStorageClient } from '@kbn/storage-adapter';
+import type { PersistedTask } from './types';
+
+export const taskStorageSettings = {
+  name: '.kibana_streams_tasks',
+  schema: {
+    properties: {
+      id: types.keyword(),
+      type: types.keyword(),
+      status: types.keyword(),
+      stream: types.keyword(),
+      space: types.keyword(),
+      created_at: types.date(),
+      // Workaround for https://github.com/elastic/kibana/issues/245974
+      task: types.object({ enabled: false }),
+    },
+  },
+} satisfies IndexStorageSettings;
+export type TaskStorageSettings = typeof taskStorageSettings;
+
+export type TaskStorageClient = IStorageClient<TaskStorageSettings, PersistedTask>;

x-pack/platform/plugins/shared/streams/server/lib/tasks/task_client.ts

@@ -0,0 +1,115 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type { KibanaRequest, Logger } from '@kbn/core/server';
+import type { TaskManagerStartContract } from '@kbn/task-manager-plugin/server';
+import { isNotFoundError, isResponseError } from '@kbn/es-errors';
+import type { TaskStorageClient } from './storage';
+import type { PersistedTask, TaskParams } from './types';
+
+interface TaskRequest<TaskType, TParams extends {}> {
+  task: Omit<PersistedTask & { type: TaskType }, 'status' | 'created_at' | 'task'>;
+  params: TParams;
+  request: KibanaRequest;
+}
+
+export class TaskClient<TaskType extends string> {
+  constructor(
+    private readonly taskManagerStart: TaskManagerStartContract,
+    private readonly storageClient: TaskStorageClient,
+    private readonly logger: Logger
+  ) {}
+
+  public async get<TPayload extends {} = {}>(id: string): Promise<PersistedTask<TPayload>> {
+    try {
+      this.logger.debug(`Getting task ${id}`);
+
+      const response = await this.storageClient.get({
+        id,
+      });
+
+      if (!response._source) {
+        // Should not happen
+        throw new Error(`Task ${id} has no source`);
+      }
+
+      return response._source as PersistedTask<TPayload>;
+    } catch (error) {
+      if (isNotFoundError(error)) {
+        return {
+          id,
+          status: 'not_started',
+          created_at: '',
+          space: '',
+          stream: '',
+          type: '',
+        };
+      }
+
+      throw error;
+    }
+  }
+
+  public async schedule<TParams extends {} = {}>({
+    task,
+    params,
+    request,
+  }: TaskRequest<TaskType, TParams>): Promise<PersistedTask> {
+    const taskDoc: PersistedTask = {
+      ...task,
+      status: 'in_progress',
+      created_at: new Date().toISOString(),
+    };
+
+    try {
+      await this.taskManagerStart.schedule(
+        {
+          id: task.id,
+          taskType: task.type,
+          params: {
+            ...params,
+            _task: {
+              ...taskDoc,
+            },
+          } satisfies TaskParams<TParams>,
+          state: {},
+          scope: ['streams'],
+          priority: 1,
+        },
+        {
+          request,
+        }
+      );
+
+      this.logger.debug(`Scheduled ${task.type} task (${task.id})`);
+      await this.update(taskDoc);
+    } catch (error) {
+      const isVersionConflict =
+        isResponseError(error) && error.message.includes('version conflict');
+      if (!isVersionConflict) {
+        throw error;
+      }
+    }
+
+    return taskDoc;
+  }
+
+  public async update<TPayload extends {} = {}>(
+    task: PersistedTask<TPayload>
+  ): Promise<PersistedTask<TPayload>> {
+    this.logger.debug(`Updating task ${task.id}`);
+
+    await this.storageClient.index({
+      id: task.id,
+      document: task,
+      // This might cause issues if there are many updates in a short time from multiple tasks running concurrently
+      refresh: true,
+    });
+
+    return task;
+  }
+}

x-pack/platform/plugins/shared/streams/server/lib/tasks/task_definitions/index.ts

@@ -0,0 +1,19 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type { TaskDefinitionRegistry } from '@kbn/task-manager-plugin/server';
+import type { GetScopedClients } from '../../../routes/types';
+
+export interface TaskContext {
+  getScopedClients: GetScopedClients;
+}
+
+export function createTaskDefinitions(taskContext: TaskContext) {
+  return {} satisfies TaskDefinitionRegistry;
+}
+
+export type StreamsTaskType = keyof ReturnType<typeof createTaskDefinitions>;

x-pack/platform/plugins/shared/streams/server/lib/tasks/task_service.ts

@@ -0,0 +1,45 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type {
+  TaskManagerSetupContract,
+  TaskManagerStartContract,
+} from '@kbn/task-manager-plugin/server';
+import type { CoreStart, Logger } from '@kbn/core/server';
+import { StorageIndexAdapter } from '@kbn/storage-adapter';
+import { TaskClient } from './task_client';
+import type { TaskStorageSettings } from './storage';
+import { taskStorageSettings } from './storage';
+import type { PersistedTask } from './types';
+import type { TaskContext } from './task_definitions';
+import { createTaskDefinitions, type StreamsTaskType } from './task_definitions';
+
+export class TaskService {
+  constructor(private readonly taskManagerSetup: TaskManagerSetupContract) {}
+
+  public registerTasks(taskContext: TaskContext) {
+    this.taskManagerSetup.registerTaskDefinitions(createTaskDefinitions(taskContext));
+  }
+
+  public async getClient(
+    coreStart: CoreStart,
+    taskManagerStart: TaskManagerStartContract,
+    logger: Logger
+  ) {
+    const storageAdapter = new StorageIndexAdapter<TaskStorageSettings, PersistedTask>(
+      coreStart.elasticsearch.client.asInternalUser,
+      logger.get('task_client', 'storage'),
+      taskStorageSettings
+    );
+
+    return new TaskClient<StreamsTaskType>(
+      taskManagerStart,
+      storageAdapter.getClient(),
+      logger.get('task_client')
+    );
+  }
+}

x-pack/platform/plugins/shared/streams/server/lib/tasks/types.ts

@@ -0,0 +1,44 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+interface PersistedTaskBase {
+  id: string;
+  type: string;
+  status: 'not_started' | 'in_progress' | 'completed' | 'failed';
+  stream: string;
+  space: string;
+  created_at: string;
+}
+
+interface NotStartedTask extends PersistedTaskBase {
+  status: 'not_started';
+}
+interface InProgressTask extends PersistedTaskBase {
+  status: 'in_progress';
+}
+interface CompletedTask<TPayload extends {}> extends PersistedTaskBase {
+  status: 'completed';
+  task: {
+    payload: TPayload;
+  };
+}
+interface FailedTask extends PersistedTaskBase {
+  status: 'failed';
+  task: {
+    error: string;
+  };
+}
+
+export type PersistedTask<TPayload extends {} = {}> =
+  | NotStartedTask
+  | InProgressTask
+  | CompletedTask<TPayload>
+  | FailedTask;
+
+export type TaskParams<TParams extends {} = {}> = TParams & {
+  _task: PersistedTask;
+};