IoT Edge Digital Twin with Drift Reconciliation

id: iot-edge-digital-twin-reactor
namespace: company.team

description: |
  Maintain a per-device digital twin from realtime MQTT telemetry: merge each
  message into a Redis-backed shadow document, detect drift between desired and
  reported state, and publish a reconcile command back to the device only when
  the reported state actually changes - so steady-state drift never spams the
  command topic.

triggers:
  - id: telemetry_stream
    type: io.kestra.plugin.mqtt.RealtimeTrigger
    description: One execution per telemetry message on any device, with no polling delay.
    server: "{{ secret('MQTT_SERVER') }}"
    clientId: kestra-digital-twin
    topic: "devices/+/telemetry"
    serdeType: JSON
    qos: 1
    disabled: true

inputs:
  - id: desired_state
    type: JSON
    displayName: Desired state
    description: Source of truth for what every device should report back.
    defaults: |
      {"power": "on", "mode": "auto", "report_interval_s": 60}

tasks:
  - id: log_telemetry
    type: io.kestra.plugin.core.log.Log
    description: Record the incoming telemetry in the execution logs for the audit trail.
    message: "Twin telemetry {{ trigger.topic }} device={{ trigger.payload.device_id }}"

  - id: read_twin
    type: io.kestra.plugin.redis.string.Get
    description: Load the current shadow document for this device, if one exists.
    url: "{{ secret('REDIS_URL') }}"
    key: "twin:{{ trigger.payload.device_id }}"
    failedOnMissing: false

  - id: compute_delta
    type: io.kestra.plugin.scripts.python.Script
    description: Merge telemetry into the shadow, diff reported against desired, and
      build the reconcile command.
    env:
      TELEMETRY: "{{ trigger.payload | toJson }}"
      PRIOR_TWIN: "{{ outputs.read_twin.value | default('{}') }}"
      DESIRED: "{{ inputs.desired_state | toJson }}"
      TOPIC: "{{ trigger.topic }}"
      EXECUTION_ID: "{{ execution.id }}"
    script: |
      import json
      import os
      from datetime import datetime, timezone

      def load(raw, fallback):
          try:
              value = json.loads(raw or "")
              return value if isinstance(value, dict) else fallback
          except Exception:
              return fallback

      telemetry = load(os.environ.get("TELEMETRY"), {})
      prior = load(os.environ.get("PRIOR_TWIN"), {})
      desired = load(os.environ.get("DESIRED"), {})

      topic = os.environ.get("TOPIC") or "unknown"
      parts = topic.split("/")
      topic_device = parts[1] if len(parts) >= 2 and parts[0] == "devices" else topic
      device_id = str(telemetry.get("device_id") or topic_device)
      is_new = not bool(prior)
      previous_reported = prior.get("reported") if isinstance(prior.get("reported"), dict) else {}
      reported = {k: v for k, v in telemetry.items() if k != "device_id"}
      changed = reported != previous_reported

      drift = []
      for field in sorted(desired):
          want = desired[field]
          have = reported.get(field)
          if have != want:
              drift.append({"field": field, "desired": want, "reported": have})
      drift_set = {row["field"]: row["desired"] for row in drift}

      now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
      twin = {
          "device_id": device_id,
          "last_seen": now,
          "previous_seen": prior.get("last_seen"),
          "reported": reported,
          "desired": desired,
          "drift_fields": [row["field"] for row in drift],
          "in_sync": not drift,
      }

      command_doc = {
          "action": "reconcile",
          "device_id": device_id,
          "set": drift_set,
          "reason": "digital-twin-drift",
          "execution_id": os.environ.get("EXECUTION_ID", ""),
          "issued_at": now,
      }

      print(
          "device {} changed={} new={} drift={}".format(
              device_id, changed, is_new, ",".join(row["field"] for row in drift) or "none"
          )
      )
      print(
          "::"
          + json.dumps(
              {
                  "outputs": {
                      "device_id": device_id,
                      "twin": twin,
                      "reported": reported,
                      "drift": drift,
                      "drift_set": drift_set,
                      "command_doc": command_doc,
                      "changed": changed,
                      "is_new": is_new,
                      "in_sync": not drift,
                      "reconcile_planned": (changed or is_new) and bool(drift) and bool(reported),
                  }
              }
          )
          + "::"
      )

  - id: write_twin
    type: io.kestra.plugin.redis.string.Set
    description: Persist the updated shadow so the next message has a baseline.
    url: "{{ secret('REDIS_URL') }}"
    key: "twin:{{ trigger.payload.device_id }}"
    value: "{{ outputs.compute_delta.twin | toJson }}"

  - id: change_gate
    type: io.kestra.plugin.core.flow.If
    description: React only when reported state changed or the device is brand new -
      steady telemetry never spams commands.
    condition: "{{ outputs.compute_delta.changed or outputs.compute_delta.is_new }}"
    then:
      - id: drift_gate
        type: io.kestra.plugin.core.flow.If
        description: Publish a reconcile command only when reported state is present,
          newly changed, and misses the desired state.
        condition: "{{ outputs.compute_delta.reconcile_planned }}"
        then:
          - id: publish_reconcile
            type: io.kestra.plugin.mqtt.Publish
            description: Send the reconcile command to the device command topic with QoS 1.
            server: "{{ secret('MQTT_SERVER') }}"
            clientId: kestra-twin-reconciler
            topic: "devices/{{ trigger.payload.device_id }}/commands"
            serdeType: JSON
            qos: 1
            from:
              action: "reconcile"
              device_id: "{{ trigger.payload.device_id }}"
              set: "{{ outputs.compute_delta.drift_set | toJson }}"
              reason: "digital-twin-drift"
              execution_id: "{{ execution.id }}"

          - id: notify_drift
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Post the drift details and the command that was sent.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": ":satellite: *Digital twin drift* for {{ outputs.compute_delta.device_id }}\n{% for row in outputs.compute_delta.drift %}• {{ row.field }}: reported {{ row.reported }} != desired {{ row.desired }}\n{% endfor %}Reconcile command published to devices/{{ outputs.compute_delta.device_id }}/commands."
              }

        else:
          - id: log_in_sync
            type: io.kestra.plugin.core.log.Log
            description: State changed but already matches desired - nothing to send.
            message: "Twin for {{ outputs.compute_delta.device_id }} changed but is in sync
              - no reconcile command needed."

    else:
      - id: log_unchanged
        type: io.kestra.plugin.core.log.Log
        description: Identical telemetry - shadow baseline kept, no downstream work.
        message: "No reported-state change for {{ outputs.compute_delta.device_id }} -
          twin refreshed silently."

outputs:
  - id: device_id
    type: STRING
    description: Device the telemetry came from.
    value: "{{ outputs.compute_delta.device_id }}"

  - id: twin
    type: JSON
    description: The updated shadow document persisted in Redis.
    value: "{{ outputs.compute_delta.twin }}"

  - id: drift
    type: JSON
    description: Fields where reported state misses the desired state.
    value: "{{ outputs.compute_delta.drift }}"

  - id: reconcile_published
    type: STRING
    description: Whether a reconcile command was sent for this message.
    value: "{{ outputs.compute_delta.reconcile_planned ? 'yes' : 'no' }}"

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when twin handling itself fails so shadow updates never fail
      silently.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": ":rotating_light: *iot-edge-digital-twin-reactor* failed for device {{ trigger.payload.device_id | default('unknown') }} in flow {{ flow.id }} (execution {{ execution.id }}) - check the execution logs."
      }