docs: add Kafka message contract

At the northbound, the Package Manager listens to Kafka topics where SONATA publishes service lifecycle requests.

Each topic maps to a dedicated handler. At the southbound, handlers interact with the Helm client, Kubernetes API, and the Registry API service.

The full message contract (payloads, validation rules, registry round trip) is in [docs/kafka-api.md](docs/kafka-api.md).

**Incoming topics:**

| SONATA                      | `helm-engine-deploy`                    | Registry API → Helm client       | Deploy a chart from the internal registry onto a K8s cluster   |

| SONATA                      | `helm-engine-activation`                | Registry API → Helm client       | Activate or deactivate a deployed service                      |

| SONATA                      | `helm-engine-termination`               | Registry API → Helm client       | Uninstall a Helm release and clean up its namespace            |

| Registry API                | `get-package-registry-incoming-channel` | Internal handler routing         | Deliver credentials or kubeconfig needed by the above handlers |

| Registry API                | `response-package-registry-outgoing-channel` | Internal handler routing    | Deliver credentials or kubeconfig needed by the above handlers |

**Outgoing topics:**

| SONATA                      | `helm-engine-deploy-result`                  | Result of the Helm chart deployment                   |

| SONATA                      | `helm-engine-activation-result`              | Result of the service activation or deactivation      |

| SONATA                      | `helm-engine-termination-result`             | Result of the Helm release termination                |

| Registry API                | `response-package-registry-outgoing-channel` | Request credentials or kubeconfig from Registry API   |

| Registry API                | `get-package-registry-incoming-channel`      | Request credentials or kubeconfig from Registry API   |

| Registry API                | `registry-store-kubernetes-config`           | Store the namespace-scoped ServiceAccount kubeconfig  |

---

}

```

The `servicePackageManager` field drives chart acquisition:

The source type is detected from the shape of `serviceRepositoryURL` and `serviceArtifactIdentifier`. The `servicePackageManager` field is required by validation but not used:

| Value        | Source type                  | Conversion applied              |

|:-------------|:-----------------------------|:--------------------------------|

| `helm`       | Helm repository              | None — used directly            |

| `oci`        | OCI registry                 | None — pulled via OCI client    |

| `docker`     | Git repo with `compose.yaml` | Docker Compose → Helm (Kompose) |

| `kubernetes` | Git repo with K8s manifests  | Kubernetes manifests → Helm (Kompose) |

| Repository URL / identifier            | Source type                  | Conversion applied              |

|:---------------------------------------|:-----------------------------|:--------------------------------|

| `http…`                                | Helm repository              | None — used directly            |

| `oci://…`                              | OCI registry                 | None — pulled via OCI client    |

| `….git` with a `.yml` identifier       | Git repo with a compose file | Docker Compose → Helm (Kompose) |

| `….git`                                | Git repo with K8s manifests  | Kubernetes manifests → Helm (Kompose) |

If the source repository is private, the handler routes to the Registry API (via `get-package-registry-incoming-channel`) to retrieve credentials before fetching the chart. For OCM-based multi-cluster deployments, it also routes through Registry API so the scheduler namespace can be resolved and embedded into the chart annotations.

If the source repository is private, the handler routes to the Registry API (via `get-package-registry-incoming-channel`) to retrieve credentials before fetching the chart. For OCM-based multi-cluster deployments (`clusterManager: "ocm"`), it also routes through Registry API, passing the scheduler namespace read from `clusterMetadata`, and the chart is rewritten into a P2Code `ManifestWork` before it is stored.

Upon success, the handler publishes a response to `helm-engine-parse-result`:

{

  "specversion": "1.0",

  "id": "21000e26-31eb-43e7-8343-88a696fd96b1",

  "source": "/process/helmEngineParseRegistryProcess",

  "source": "",

  "type": "helm-engine-parse-result",

  "kogitoprocrefid": "7aa45064-f4a7-4a25-a7a2-667afb07c3d1",

  "data": {

    "serviceOrderId": "2c4fecbc-c436-4f8c-a5db-4a1159ebad2a",

    "serviceId": "790daa4d-7f56-4575-8b64-c558cf518bed",

    "helmEngineOrderStatus": "success",

    "helmEngineOrderMessage": "Helm charts were compiled and pushed successfully",

    "helmEngineOrderStatus": "SUCCESS",

    "helmEngineOrderMessage": "Helm charts were parsed successfully",

    "clusters": [

      {

        "locationId": "",

        "locationId": "590037ca-c319-41cd-a5fd-624028e60933",

        "numberOfK8sNodes": 1

      }

    ]

The `activateService` flag controls the initial state of the deployment. When `false`, the release is installed and then immediately uninstalled with history kept (`helm uninstall --keep-history`), leaving the namespace intact but the service inactive and restorable.

The deployment namespace defaults to the `serviceId`. For OCM-based deployments, the namespace is overridden with the scheduler namespace extracted from the chart annotations during the parse phase.

The deployment namespace defaults to the `serviceId`. For OCM-based deployments, the Registry API returns the scheduler namespace in its kubeconfig reply (`schedulerNamespace`), and that namespace is used instead.

Upon success, the handler publishes to `helm-engine-deploy-result`:

{

  "specversion": "1.0",

  "id": "21000e26-31eb-43e7-8343-88a696fd96b1",

  "source": "/process/helmEngineDeploymentProcess",

  "source": "",

  "type": "helm-engine-deploy-result",

  "kogitoprocrefid": "7aa45064-f4a7-4a25-a7a2-667afb07c3d1",

  "data": {

    "serviceOrderId": "2c4fecbc-c436-4f8c-a5db-4a1159ebad2a",

    "serviceId": "790daa4d-7f56-4575-8b64-c558cf518bed",

    "helmEngineOrderStatus": "success",

    "helmEngineOrderStatus": "SUCCESS",

    "helmEngineOrderMessage": "Helm charts were deployed successfully",

    "state": "ACTIVE",

    "releaseName": "release-790daa4d",

}

```

For standard deployments, the namespace is deleted along with the release. For OCM-based deployments (where the scheduler namespace is set in chart annotations), the namespace is preserved because it is managed by the OCM scheduler, not by the Package Manager.

For standard deployments, the namespace is deleted along with the release. For OCM-based deployments (where the Registry API returns a scheduler namespace), the namespace is preserved because it is managed by the OCM scheduler, not by the Package Manager.

The response is published to `helm-engine-termination-result`:

{

  "specversion": "1.0",

  "id": "21000e26-31eb-43e7-8343-88a696fd96b1",

  "source": "/process/helmEngineTerminationProcess",

  "source": "",

  "type": "helm-engine-termination-result",

  "kogitoprocrefid": "7aa45064-f4a7-4a25-a7a2-667afb07c3d1",

  "data": {

    "serviceOrderId": "2c4fecbc-c436-4f8c-a5db-4a1159ebad2a",

    "serviceId": "790daa4d-7f56-4575-8b64-c558cf518bed",

    "helmEngineOrderStatus": "success",

    "helmEngineOrderStatus": "SUCCESS",

    "helmEngineOrderMessage": "Helm release termination was successful"

  }

}

| `TypeActivation` | `HandleActivation`  |

| `TypeTermination`| `HandleTermination` |

The two credential message structures are shown below.

**Repository / registry credentials** (used for chart fetch in the parse flow):

```go

type RegistryURLMessage struct {

    Type               string                                     `json:"type"`

    KogitoServiceOrder kogito.RequestData[datamodel.ServiceOrder] `json:"kogitoServiceOrder"`

    RepoCredentials    RepoCredentials                            `json:"repoCredentials"`

}

type RepoCredentials struct {

    Username string `json:"username"`

    Password string `json:"password"`

}

```

**Image pull secret + kubeconfig** (used for deploy, activate, and terminate flows):

```go

type PullSecretMessage struct {

    Type              string                                          `json:"type"`

    HelmDeployRequest kogito.RequestData[datamodel.HelmDeployRequest] `json:"helmDeployRequest"`

    ImagePullSecret   ImagePullSecret                                 `json:"imagePullSecret"`

}

type ImagePullSecret struct {

    URL      string `json:"url"`

    Username string `json:"username"`

    Password string `json:"password"`

}

```

The request and reply payloads are documented in [docs/kafka-api.md](docs/kafka-api.md#registry-requests).

---

# Kafka message contract

## Topics

All topic names come from `service.yml` (`kafka.topics.<key>.{request,response}`), resolved in

`main.go` (`resolveKafkaTopics`). `GetTopics(key)` returns `(request, response)`, used as `(In, Out)`.

| Topic (default) | Dir | Bound as | Purpose |

|---|---|---|---|

| `helm-engine-parse` | in | `parse.request` | Fetch/convert a chart and store it in the internal OCI registry. |

| `helm-engine-parse-result` | out | `parse.response` | Parse outcome. |

| `helm-engine-deploy` | in | `deploy.request` | Install a stored chart on a cluster. |

| `helm-engine-deploy-result` | out | `deploy.response` | Deploy outcome. |

| `helm-engine-activation` | in | `activate.request` | Restore a deactivated release (`helm rollback`). |

| `helm-engine-activation-result` | out | `activate.response` | Activation outcome. |

| `helm-engine-termination` | in | `terminate.request` | Uninstall a release. |

| `helm-engine-termination-result` | out | `terminate.response` | Termination outcome. |

| `get-package-registry-incoming-channel` | out | `registry.response` | Credential / kubeconfig requests to the package-registry. |

| `response-package-registry-outgoing-channel` | in | `registry.request` | The package-registry's replies; every operation resumes here. |

| `registry-store-kubernetes-config` | out | `store_kubeconfig.request` | Namespace-scoped ServiceAccount kubeconfig, stored by the registry. |

The `registry` entry swaps `request`/`response` on purpose: the names are written from the

package-registry's point of view (its "outgoing channel" is this service's input).

`kafka.topics.store` in `service.yml` is not read by the code.

Topics that don't exist yet are created with one partition (`EnsureStreamExists(topic, 1)` in

base.go). All messages are emitted with an empty key.

## Request → result flow

Every operation is two Kafka hops. This service holds no credentials and no state between them.

1. A request lands on its input topic (`parse`, `deploy`, `activate`, `terminate`).

2. The handler validates it and, if it needs credentials, publishes a `RegistryURLRequest` or

   `KubeconfigRequest` to `get-package-registry-incoming-channel` and **returns**. The request

   carries the original order/operation (`serviceOrderRequest` / `serviceOperationRequest`),

   which is the resume state.

3. The package-registry resolves the credentials and replies on

   `response-package-registry-outgoing-channel`, echoing the original order/operation back.

4. `ReceiveRegistryCredentials` (`ops/registryHandler.go`) routes on `data.type`

   (`parsing` / `deployment` / `activation` / `termination`) to `HandleParsing` /

   `HandleDeployment` / `HandleActivation` / `HandleTermination`, which rebuild the operation

   from the echoed payload. Any replica can handle the reply.

5. The result is published on the matching `*-result` topic.

Exception: a parse of a **public** repository with **no** `clusterMetadata` skips the registry

and parses directly (steps 2–4 don't happen).

## Kogito envelope

The four request topics use the standard Kogito process envelope

(`base.go/pkg/adapters/kogito.RequestData[T]`):

```json

{ "kogitoprocinstanceid": "7aa45064-f4a7-4a25-a7a2-667afb07c3d1", "source": "/process/x", "data": {} }

```

- `kogitoprocinstanceid` and `source` must each be at least 4 characters

  (`kogito.RequestData.Validate()`), then the payload's own `Validate()` runs.

- On a decode or validation failure the handler replies on its own `*-result` topic with

  `helmEngineOrderStatus: "FAILURE"` and the error in `helmEngineOrderMessage`.

`response-package-registry-outgoing-channel` is not Kogito-wrapped; see

[Registry replies](#registry-replies).

Inbound Kafka headers (including `Authorization`, which the registry requires, and the

OpenTelemetry trace context) are forwarded on every message this service emits.

## Requests

### `ServiceOrder`: `helm-engine-parse`

`datamodel/serviceOrder.go`

```json

{

  "kogitoprocinstanceid": "proc-123",

  "source": "/process/serviceDeploymentProcess",

  "data": {

    "serviceOrderId": "so-1234",

    "serviceId": "svc-5678",

    "servicePackageManager": "helm",

    "serviceRepositoryURL": "https://charts.example.org",

    "serviceArtifactIdentifier": "nginx",

    "serviceArtifactVersion": "1.2.3",

    "servicePackageValuesYml": "<base64 values.yaml>",

    "clusterManager": "",

    "clusterMetadata": ""

  }

}

```

| Field | Validation / meaning |

|---|---|

| `serviceOrderId` | not validated; echoed in the result |

| `serviceId` | at least 5 characters; the storage key (`helm-engine/<serviceId>:0.0.1`) |

| `servicePackageManager` | required, but **not used**: the repo kind is detected from the URL (below) |

| `serviceRepositoryURL` | required; host must be a valid IP/hostname (`commons.ValidateIPOrHost`) |

| `serviceArtifactIdentifier` | required; chart name, or compose file name for Docker repos |

| `serviceArtifactVersion` | optional |

| `servicePackageValuesYml` | optional, base64 |

| `clusterManager` | `"ocm"` selects the P2Code / ManifestWork adapter |

| `clusterMetadata` | optional, base64 `P2CodeSchedulingManifest` YAML; its `metadata.namespace` becomes the `schedulerNamespace` sent to the registry |

Repo kind (`detectRepoKind`, `ops/operator.go`), from the URL and identifier alone:

| Shape | Kind | Acquisition |

|---|---|---|

| `oci://…` | OCI | pull |

| `….git` + `.yml` identifier | DOCKER | git clone + Kompose |

| `….git` | KUBERNETES | git clone + Kompose |

| `http…` | HELM | pull from index |

### `HelmDeployRequest`: `helm-engine-deploy`

`datamodel/helmDeploy.go`

```json

{

  "kogitoprocinstanceid": "proc-123",

  "source": "/process/serviceDeploymentProcess",

  "data": {

    "serviceOrderId": "so-1234",

    "serviceId": "svc-5678",

    "activateService": true,

    "k8sServiceId": "2e5b1f9a-6c3d-4a2e-8b0a-9f7c1d4e5a6b",

    "servicePackageValuesYml": "<base64 values.yaml>",

    "requestedCompletionDate": "2026-09-30T12:00:00Z"

  }

}

```

| Field | Validation / meaning |

|---|---|

| `serviceOrderId` | not validated; echoed in the result |

| `serviceId` | at least 4 characters; selects the stored chart and the release name |

| `activateService` | `false`: install, then `helm uninstall --keep-history`, so the release can be activated later |

| `k8sServiceId` | at least 4 characters; the registry key for the cluster kubeconfig |

| `servicePackageValuesYml` | optional, base64 |

| `requestedCompletionDate` | not validated here, but **required by the registry reply's** `Validate()`: a deploy without it fails after the round trip |

Before contacting the registry, the handler reads the stored chart for image-pull-secret names

and sends them as `secretNames`.

### `ActivationRequest`: `helm-engine-activation`

`datamodel/activation.go`

```json

{

  "kogitoprocinstanceid": "proc-123",

  "source": "/process/serviceActivationProcess",

  "data": {

    "serviceOrderId": "so-1234",

    "serviceId": "svc-5678",

    "activateService": true,

    "k8sServiceId": "2e5b1f9a-6c3d-4a2e-8b0a-9f7c1d4e5a6b"

  }

}

```

| Field | Validation |

|---|---|

| `serviceOrderId` | at least 4 characters |

| `serviceId` | at least 4 characters |

| `activateService` | forwarded to the registry |

| `k8sServiceId` | not validated; the registry key for the kubeconfig |

### `TerminationRequest`: `helm-engine-termination`

`datamodel/termination.go`

```json

{

  "kogitoprocinstanceid": "proc-123",

  "source": "/process/serviceTerminationProcess",

  "data": {

    "serviceOrderId": "so-1234",

    "serviceId": "svc-5678",

    "k8sServiceId": "2e5b1f9a-6c3d-4a2e-8b0a-9f7c1d4e5a6b"

  }

}

```

| Field | Validation |

|---|---|

| `serviceOrderId` | at least 4 characters |

| `serviceId` | at least 4 characters |

| `k8sServiceId` | not validated; the registry key for the kubeconfig |

## Registry requests

Published to `get-package-registry-incoming-channel` (`registryapi/requests.go`,

`registryapi/helpers.go`). No Kogito envelope. The registry picks the Java type from `type`.

### `RegistryURLRequest` (`type: parsing`)

Sent for a private repository, or whenever `clusterMetadata` yields a scheduler namespace.

```json

{

  "type": "parsing",

  "serviceOrderRequest": {

    "kogitoprocinstanceid": "proc-123",

    "serviceOrder": { "serviceId": "svc-5678", "serviceRepositoryURL": "…" }

  },

  "url": "charts.example.org",

  "schedulerNamespace": ""

}

```

| Field | Meaning |

|---|---|

| `serviceOrderRequest` | the full inbound `ServiceOrder` plus its process ID; echoed back |

| `url` | the repo URL used as the Vault lookup key; empty for public repos |

| `schedulerNamespace` | from `clusterMetadata`; empty when not OCM |

### `KubeconfigRequest` (`type: deployment | activation | termination`)

```json

{

  "type": "deployment",

  "serviceOperationRequest": {

    "kogitoprocinstanceid": "proc-123",

    "serviceOrderId": "so-1234",

    "serviceId": "svc-5678",

    "activateService": true,

    "requestedCompletionDate": "2026-09-30T12:00:00Z",

    "servicePackageValuesYml": "<base64 values.yaml>"

  },

  "kubernetesServiceID": "2e5b1f9a-6c3d-4a2e-8b0a-9f7c1d4e5a6b",

  "secretNames": ["regcred"]

}

```

| Field | Meaning |

|---|---|

| `serviceOperationRequest` | the inbound request plus its process ID; echoed back. Termination always sends `activateService: false` |

| `kubernetesServiceID` | the inbound `k8sServiceId` |

| `secretNames` | image-pull-secret names from the chart (deployment only; `null` otherwise) |

## Registry replies

Consumed from `response-package-registry-outgoing-channel`. The shape mirrors the registry's

`PackageRegistrySecretResponseMessage` / `KubeConfigResponseMessage`:

```json

{

  "id": "svc-5678",

  "timestamp": "…",

  "status": "SUCCESS",

  "message": "",

  "data": { "type": "deployment", "serviceOperationRequest": { "…": "echoed" } }

}

```

The process ID is **not** on the envelope: it is read from the echoed

`data.serviceOrderRequest.kogitoprocinstanceid` or

`data.serviceOperationRequest.kogitoprocinstanceid`.

`status` is not checked. A `deployment` / `activation` / `termination` failure reply has no

`kubeConfigAsBase64`, so it fails the decode validation below and is reported as a failure. A

`parsing` failure reply still passes validation and parsing continues with empty credentials, so

it fails later, at the private-repo fetch.

### `RegistryURLData` (`type: parsing`)

`registryapi/messages.go`

| Field | Meaning |

|---|---|

| `serviceOrderRequest` | echoed; must pass `ServiceOrder.Validate()` again |

| `repoCredentials` | `{ "username", "password" }` for the private repo; absent for public |

### `KubeconfigData` (`type: deployment | activation | termination`)

| Field | Validation / meaning |

|---|---|

| `serviceOperationRequest` | echoed; `serviceId` must be at least 5 characters |

| `kubeConfigAsBase64` | required; the cluster kubeconfig |

| `imagePullSecrets` | `[{ "secretName", "url", "username", "password" }]`, created in the target namespace (non-OCM deploys) |

| `schedulerNamespace` | non-empty means OCM: used as the target namespace instead of `serviceId`, and selects `OCMStrategy` |

| `requestedCompletionDate` | (inside `serviceOperationRequest`) required when `type` is `deployment` |

`kubeConfigAsBase64` and `servicePackageValuesYml` are base64-decoded after validation

(`KubeconfigMessageDecoder`).

## SA kubeconfig store

After a successful non-OCM deploy (`DefaultHelmStrategy.PostInstall`), a namespace-scoped

ServiceAccount kubeconfig (read-only on pods/services/configmaps) is published to

`registry-store-kubernetes-config`:

```json

{

  "data": {

    "serviceId": "svc-5678",

    "kubeconfigScope": "NAMESPACE_SA",

    "kubeConfig": "<base64 kubeconfig>"

  }

}

```

The registry's store result is not consumed by this service.

## Results

Every `*-result` message is a Kogito response envelope (`kogito.ResponseData[T]`, built by

`createKogitoResponse` in `ops/kogito_ops.go`):

| Field | Value |

|---|---|

| `specversion` | `"1.0"` |

| `id` | a fresh UUID per message |

| `source` | always `""` |

| `type` | the output topic name |

| `kogitoprocrefid` | the originating request's `kogitoprocinstanceid` |

| `data` | per-operation payload, below |

`helmEngineOrderStatus` is `"SUCCESS"` or `"FAILURE"` in every payload.

### `ServiceOrderResponse`: `helm-engine-parse-result`

| Field | Meaning |

|---|---|

| `serviceOrderId`, `serviceId` | echoed |

| `helmEngineOrderStatus` | `SUCCESS` / `FAILURE` |

| `helmEngineOrderMessage` | `"Helm charts were parsed successfully"` or the error |

| `clusters` | on success, one fixed entry `{ "locationId": "590037ca-c319-41cd-a5fd-624028e60933", "numberOfK8sNodes": 1 }`; `[]` on failure |

### `HelmDeployResponse`: `helm-engine-deploy-result` and `helm-engine-activation-result`

Deploy and activation share this payload.

| Field | Meaning |

|---|---|

| `serviceOrderId`, `serviceId` | echoed |

| `helmEngineOrderStatus` | `SUCCESS` / `FAILURE` |

| `helmEngineOrderMessage` | `"Helm charts were deployed successfully"` or a failure message |

| `state` | `ACTIVE`, or `INACTIVE` when deployed with `activateService: false` (also on failures) |

| `releaseName` | derived from `serviceId` (`GenerateReleaseName`) |

| `k8sIpAddress` | the cluster address |

| `serviceNamespaces` | target namespace(s) |

| `serviceEndpoints` | `[{ "ip", "port" }]` |

| `namespaceResourceQuota` | optional |

| `defaultValuesYaml` | optional; the chart's default values |

### `TerminationResponse`: `helm-engine-termination-result`

| Field | Meaning |

|---|---|

| `serviceOrderId`, `serviceId` | echoed |

| `helmEngineOrderStatus` | `SUCCESS` / `FAILURE` |

| `helmEngineOrderMessage` | `"Helm release termination was successful"` or a failure message |

On OCM termination the namespace is kept (OCM owns it); otherwise it is deleted.

## Failure reporting quirks

- A registry reply that fails to decode is reported by `logAndHandleRegistryError` as a

  **parse-shaped** `ServiceOrderResponse`, published on the result topic of the reply's

  `type` (for example a parse-shaped message on `helm-engine-deploy-result`).

- If publishing the activation `KubeconfigRequest` fails, no result is emitted

  (`ActivationRequestHandler` only records the error on the trace span).

- If the registry never replies, nothing is emitted: this service has no timeout on the

  round trip.