Тип действия batch_playbook в политике

id: evo.vulnerability.policy.batch
type: calculation_schema
name: Политика создания задач на устранение уязвимостей с действием типа batch_playbook
tags: [vulnerabilities, policy, automation]
version: 1.0.0
status: stable
author: John Doe <johndoe@example.com>

policies:
  - id: create_task_if_vulnerability_is_critical
    name: Создать задачу, если уязвимость критическая
    entity: vulnerabilities.rvision.ru/Vulnerability
    conditions:
      - type: expression
        expression: 'equalText({field:criticalityLevel}, "Критический")'
    actions:
      - type: batch_playbook
        playbook: create_tasks_from_critical_vulnerabilities
        msgTimeout: 300000

id: create_tasks_from_critical_vulnerabilities
type: script
name: Создать задачи из критических уязвимостей
tags: [vulnerabilities, policy, automation]
version: 1.0.0
status: stable
author: John Doe <johndoe@example.com>

source: !deno |

  import { type Context, executeHandler } from '@expertise/policy_batch_action_handler_executor';
  import { cms } from '@playbooks/sdk';
  import type { PolicyChunkEventResult } from 'npm:@evo.cms/policy-module.contract@1.11.83';
  import type { CmsObject } from 'npm:@evo.cms/policy-module.contract@1.11.83/dist/evo/object_mutation/v1/object_mutation';

  export async function createTasksFromCriticalVulnerabilities(cmsObjects: CmsObject[]): Promise<PolicyChunkEventResult[]> {
    await cms.mass_create({
      domainId: 'tasks.rvision.ru',
      entityId: 'Task',
      valuesList: cmsObjects.map((cmsObject) => {
        const values = JSON.parse(cmsObject.values!);

        return JSON.stringify({
          taskName: "Устранить уязвимость ".concat(values.name),
        });
      }),
    });

    return cmsObjects.map((cmsObject) => ({ policyChunkResult: { id: cmsObject.id } }));
  }

  export async function main(): Promise<void> {
    const context: Context = JSON.parse(get_context());

    await executeHandler({
      handler: createTasksFromCriticalVulnerabilities,
      batchSize: 1000,
      context,
    });
  }

id: policy_batch_action_handler_executor
type: script
name: Массовая обработка объектов политики с помощью NATS
tags: [policy, automation]
version: 1.0.0
status: stable
author: John Doe <johndoe@example.com>

source: !deno |

  import type { ConsumerMessages } from 'jsr:@nats-io/jetstream@3.1.0';
  import { jetstream, jetstreamManager } from 'jsr:@nats-io/jetstream@3.1.0';
  import { headers } from 'jsr:@nats-io/nats-core@3.1.0';
  import { connect } from 'jsr:@nats-io/transport-deno@3.1.0';
  import process from 'node:process';
  import {
    PolicyChunkEvent,
    PolicyChunkEventFinish,
    PolicyChunkEventResult,
  } from 'npm:@evo.cms/policy-module.contract@1.11.83';
  import type { CmsObject } from 'npm:@evo.cms/policy-module.contract@1.11.83/dist/evo/object_mutation/v1/object_mutation';

  async function getCmsObjects(messages: ConsumerMessages): Promise<CmsObject[]> {
    const cmsObjects: CmsObject[] = [];

    for await (const message of messages) {
      const policyChunkEvent = PolicyChunkEvent.decode(message.data);

      if (policyChunkEvent.policyChunk && policyChunkEvent.policyChunk.object) {
        cmsObjects.push(policyChunkEvent.policyChunk.object);
      }

      message.ack();
    }

    return cmsObjects;
  }

  export type Context = {
    config: {
      readSubject: string;
      stream: string;
      writeSubject: string;
    };
  };

  export type ExecuteHandlerInput = {
    handler: (cmsObjects: CmsObject[]) => Promise<PolicyChunkEventResult[]>;
    batchSize: number;
    context: Context;
  };

  export async function executeHandler(executeHandlerInput: ExecuteHandlerInput): Promise<void> {
    const nc = await connect({
      servers: [`${process.env.NATS_ADDR}:${process.env.NATS_PORT}`],
    });
    const js = jetstream(nc);
    const jsm = await jetstreamManager(nc);

    const ephemeralConsumerInfo = await jsm.consumers.add(executeHandlerInput.context.config.stream, {
      filter_subject: executeHandlerInput.context.config.readSubject,
      deliver_policy: 'all',
      ack_policy: 'explicit',
    });
    const ephemeralConsumer = await js.consumers.get(
      executeHandlerInput.context.config.stream,
      ephemeralConsumerInfo.name,
    );

    while (true) {
      const messages = await ephemeralConsumer.fetch({ max_messages: executeHandlerInput.batchSize });
      const cmsObjects = await getCmsObjects(messages);

      if (cmsObjects.length == 0) {
        console.log('Стрим NATS полностью прочитан. Завершение чтения...');
        break;
      }

      const policyChunkEventResults = await executeHandlerInput.handler(cmsObjects);

      await Promise.all(
        policyChunkEventResults.map((policyChunkEventResult) =>
          js.publish(
            executeHandlerInput.context.config.writeSubject,
            PolicyChunkEventResult.encode(policyChunkEventResult).finish(),
          ),
        ),
      );
    }

    const h = headers();
    h.set('X-Finish-Event', 'true');

    await js.publish(executeHandlerInput.context.config.writeSubject, PolicyChunkEventFinish.encode({}).finish(), {
      headers: h,
    });
    await nc.close();
  }