feat(engine): add private NDC nodes and ontology bridge

2026-07-16 02:23:45 +03:00
parent 569b8762e6
commit 59e9c92415
25 changed files with 10450 additions and 12 deletions
@@ -0,0 +1,105 @@
+import type {
+  IExecuteFunctions,
+  ILoadOptionsFunctions,
+  INodeExecutionData,
+  INodeType,
+  INodeTypeDescription,
+} from 'n8n-workflow';
+import { NodeConnectionTypes, NodeOperationError } from 'n8n-workflow';
+
+import {
+  DATA_PRODUCT_BASE_PATH,
+  DATA_PLANE_DEFAULT_BASE_URL,
+  DATA_PLANE_BASE_URL_ENV,
+  DATA_PRODUCT_WRITER_CATALOG_PATH,
+  NDC_DATA_PRODUCT_WRITER_CREDENTIAL,
+  NDC_NODE_ICON,
+} from '../shared/constants';
+import { buildPublishPayload, encodeIdentifierPath } from '../shared/contracts';
+import { safeHttpError, safeInputError } from '../shared/errors';
+import { loadDataProductOptions, ndcRequest, serviceBaseUrl } from '../shared/http';
+
+export class NdcDataProductPublish implements INodeType {
+  description: INodeTypeDescription = {
+    displayName: 'NDC Data Product Publish',
+    name: 'ndcDataProductPublish',
+    icon: NDC_NODE_ICON,
+    group: ['output'],
+    version: 1,
+    subtitle: '={{$parameter["dataProductId"]}}',
+    description: 'Publish canonical facts through a scoped NDC Data Product grant',
+    defaults: { name: 'NDC Data Product Publish' },
+    inputs: [NodeConnectionTypes.Main],
+    outputs: [NodeConnectionTypes.Main],
+    credentials: [{ name: NDC_DATA_PRODUCT_WRITER_CREDENTIAL, required: true }],
+    properties: [
+      {
+        displayName: 'Data Product Name or ID',
+        name: 'dataProductId',
+        type: 'options',
+        typeOptions: { loadOptionsMethod: 'getDataProducts' },
+        default: '',
+        required: true,
+      },
+      {
+        displayName: 'Batch Sequence',
+        name: 'sequence',
+        type: 'number',
+        typeOptions: { minValue: 0, numberStepSize: 1 },
+        default: 0,
+        required: true,
+        description: 'Sequence number when one execution publishes several batches',
+      },
+    ],
+  };
+
+  methods = {
+    loadOptions: {
+      async getDataProducts(this: ILoadOptionsFunctions) {
+        return loadDataProductOptions(
+          this,
+          NDC_DATA_PRODUCT_WRITER_CREDENTIAL,
+          DATA_PLANE_BASE_URL_ENV,
+          DATA_PLANE_DEFAULT_BASE_URL,
+          DATA_PRODUCT_WRITER_CATALOG_PATH,
+        );
+      },
+    },
+  };
+
+  async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
+    const inputItems = this.getInputData();
+    const dataProductId = this.getNodeParameter('dataProductId', 0) as string;
+    const sequence = this.getNodeParameter('sequence', 0, 0) as number;
+    let body;
+    let encodedProductId;
+    try {
+      body = buildPublishPayload(
+        inputItems,
+        this.getExecutionId(),
+        String(this.getWorkflow().id ?? 'workflow'),
+        this.getNode().id,
+        dataProductId,
+        sequence,
+      );
+      encodedProductId = encodeIdentifierPath(dataProductId, 'dataProductId');
+    } catch (error) {
+      throw new NodeOperationError(this.getNode(), safeInputError(error, 'ndc_data_product_publish_input_invalid'));
+    }
+
+    try {
+      const baseUrl = serviceBaseUrl(DATA_PLANE_BASE_URL_ENV, DATA_PLANE_DEFAULT_BASE_URL);
+      const response = await ndcRequest(this, NDC_DATA_PRODUCT_WRITER_CREDENTIAL, {
+        method: 'POST',
+        url: `${baseUrl}${DATA_PRODUCT_BASE_PATH}/${encodedProductId}/publish`,
+        body,
+      });
+      return [[{
+        json: response as INodeExecutionData['json'],
+        pairedItem: inputItems.map((_item, index) => ({ item: index })),
+      }]];
+    } catch (error) {
+      throw new NodeOperationError(this.getNode(), safeHttpError(error, 'ndc_data_product_publish_failed'));
+    }
+  }
+}