Flow

Table of Contents

Getting Started

• Overview · Learning Roadmap · Core Concepts · Quick Start · End-to-End Tutorial · Screen Layout

Screen-by-Screen Guide

• List Screen · Edit Screen (Visual Canvas) · Execution History Screen

Message & Node Reference

• Message Structure · Payload Examples by Trigger · Node Catalog
• Triggers · Filters · Transforms
• Actions — Integration/Storage · Asset Event Publishing · Domain CRUD
• Edge · External Integration · Flow Control
• Detailed Node Option Reference · Writing Script Nodes · JS Runtime Specification · Memory-Safe Coding Patterns · Mobile Push Notification Channel · Automatic External Auth Token Refresh

Examples & Patterns

• Example Flows (17) · Operational Patterns (10 Recipes) · Use Cases

Operations

• General Workflow · Import/Export · Error Handling · Operational Diagnostics · Permissions

Production Operations

• Frequently Asked Questions (FAQ) · Operational Best Practices · Security & Sensitive Data Handling
• Flow Metrics and Alarms · Message Processing Semantics and Backpressure · What You Will See in Execution History
• Cluster & HA Behavior · End-to-End Tracing · Graph Pattern Catalog · Flow Testing Best Practices
• Performance Limits and Tuning · Auditing & History Tracking · New Flow Deployment Checklist · Deployment Strategies (Canary/Blue-Green/A·B) · Emergency Response Procedures
• Node Quick Configuration Reference · Flow REST API · Webhook Trigger · OPC/PLC Industrial Integration Patterns
• External System Integration Cookbook · Data Transformation Cookbook · Reusable Script Collection

Troubleshooting

• Step-by-Step Debugging Guide · Common Problems

Other

• Glossary · Version Notes · Related Screens

---

Overview

Flow is an automation menu where you define data integration with external systems (MES/ERP/SCADA, etc.) and the automatic creation and updating of domain objects (tags, assets, work orders, workers, and so on) as a visual graph. Operators can design, deploy, and operate automation scenarios directly, without coding.

You build data processing pipelines by dragging and dropping roughly 100 node types onto the canvas and connecting them with wires.

Path: Left menu > Automation > Flow

---

Learning Roadmap — Recommended Entry Points by Role

This manual is a comprehensive guide of over 4,000 lines. Start from the sections that match your role and goal.

👶 First-time users (build your first flow within an hour)

1. Core Concepts — 5 min
2. Quick Start — 5 min
3. End-to-End Tutorial — 30 min (follow steps 1–10)
4. Screen Layout + Edit Screen — 10 min
5. Work through Example Flows 1, 2, 3 — 10 min

→ Your first flow is deployed. After that, search the Node Catalog for whatever node you need.

🧑‍🏭 Field operators (writing automation scenarios)

1. Payload Examples by Trigger — understand the actual data structures
2. Detailed Node Option Reference — learn the options of frequently used nodes
3. Data Transformation Cookbook — copy common transformation patterns
4. Graph Pattern Catalog — pick a wiring pattern
5. Example Flows (17) — reference finished flows by scenario

🛠 System administrators (operations, tuning, incident response)

1. Flow Metrics and Alarms — which indicators to watch
2. What You Will See in Execution History — interpreting diagnostic logs
3. Cluster & HA Behavior — understanding multi-node environments
4. End-to-End Tracing — tracking a problem message
5. Performance Limits and Tuning + Emergency Response Procedures

🔌 Developers & integration engineers (external system integration)

1. External System Integration Cookbook — Slack/Teams/Jira/SAP examples
2. Flow REST API — manipulate flows programmatically
3. Webhook Trigger — fire flows from outside
4. OPC/PLC Industrial Integration Patterns — shop-floor scenarios
5. JS Runtime Specification + Reusable Script Collection

🔐 Security officers (auditing & authentication)

1. Security & Sensitive Data Handling — storing credentials
2. Automatic External Auth Token Refresh — operating OAuth2 tokens
3. Permissions — which actions each role can perform
4. Auditing & History Tracking — retaining change/execution history

📚 Quick reference (for experienced users)

What you are looking for	Section
Node IDs with one-line descriptions	Node Catalog
Node option defaults	Detailed Node Option Reference
Message JSON examples	Payload Examples by Trigger
Ready-to-use scripts	Reusable Script Collection · Data Transformation Cookbook
API call curl commands	Flow REST API
Interpreting execution history	What You Will See in Execution History
Troubleshooting guide	Step-by-Step Debugging Guide · Common Problems
Quick option tables	Node Quick Configuration Reference

All section links are anchors within this document. Searching by keyword with Ctrl+F also works well.

---

Core Concepts

Term	Description
Flow	A single automation workflow made up of nodes and wires (relations). It is a directed graph with entry points, and you control whether it is active with the Deploy/Undeploy toggle
Node	The unit that receives a message, processes it, and passes it on to the next node. Nodes fall into 7 categories (trigger, filter, transform, action, external integration, flow control, edge)
Relation	The label on the wire that goes from a node output to the next node. Nodes branch by directly assigning labels such as SUCCESS/FAILURE/TRUE/FALSE/MATCH/NO_MATCH/DEFAULT/THROTTLED/EXHAUSTED. All labels are standardized in uppercase
Message	The payload flowing through the flow. It contains type (classification), originator (subject entity), data (body), and metadata (context)
Trigger	The node that serves as a flow's starting point. There are three kinds: domain events (tag points, alarms, asset events, etc.), external entry (webhook, MQTT, external DB), and time (schedule)

---

Quick Start

The fastest way to get comfortable with Flow is these four steps.

1. On the List screen, click the 새 플로우 button to create an empty flow (enter only a name and description).
2. On the Edit screen, drag a trigger node (for example flow_on_tag_alarm) → a filter → an action (for example flow_send_email) from the left palette in order, and connect the nodes with wires.
3. Click each node, enter its options in the inspector on the right, then use Save → Test Run at the top right to verify the behavior once.
4. Turn on the Deploy toggle at the top, and the flow will run automatically whenever a trigger event arrives. You can check the results in the live debug panel and the execution history screen.

For wiring patterns for automation scenarios, see Example Flows and Use Cases in this document. For a step-by-step tutorial that builds a real flow from start to finish, follow the End-to-End Tutorial.

---

End-to-End Tutorial

A step-by-step tutorial that builds one production-ready automation flow from start to finish. Scenario: when a motor's temperature exceeds 80 °C, automatically issue an emergency maintenance work order and notify the responsible person by email.

Step 1 — Create a new flow

1. Click Automation > Flow in the left menu → open the list screen
2. Click the New Flow button at the top right
3. Enter the following information and click OK

Item	Value
Name	모터 과열 자동 정비 발행
Description	[자동화] 모터 자산의 온도 80°C 초과 시 긴급 정비 작업지시 + 이메일. 담당: ops@example.com

Once the flow is created, the edit screen opens automatically with an empty canvas.

Step 2 — Place the trigger node

Receive motor temperature data from asset telemetry events.

1. Expand the Trigger category in the left palette
2. Drag the flow_on_tag_point node onto the canvas
3. Click the node → enter the following options in the inspector on the right

Option	Value
Display name	태그 포인트 인입
tag_id_pattern	MOTOR-*.TEMP

Passing only the motor temperature tag with tag_id_pattern greatly reduces downstream processing volume. Other tags are treated as SKIPPED and are not counted.

Step 3 — Threshold filter

Add a script filter so that only messages above 80 °C pass.

1. Drag flow_script_filter from the Filter category
2. Wire the trigger node's output port to the new filter node's input port
3. Enter the following in the inspector

Option	Value
Display name	임계값 필터 (80°C 초과)
language	JS
script	data.value > 80

Step 4 — Change domain (tag → asset)

Since alarms and work orders are naturally issued per asset, convert the message's originator from the tag to its parent asset.

1. Drag flow_change_originator from the Transform category
2. Wire from the filter node's TRUE output
3. Inspector:

Option	Value
Display name	Tag → Asset 변경
entity_type	Asset
id_field	metadata.asset_id

metadata.asset_id is filled in automatically on tag point ingestion. If it is absent from the message, you can also extract the prefix portion of the tag ID (for example MOTOR-001.TEMP → MOTOR-001) with a script.

Step 5 — Issue the work order

Automatically create an emergency maintenance work order.

1. Drag flow_create_work_order from the Action — Domain CRUD category
2. Wire from the transform node's SUCCESS output
3. Inspector:

Option	Value
Display name	긴급 정비 작업지시 생성
asset_id_field	originator.id
title_field	(static value) 긴급 점검 — 모터 과열
master_id_field	(optional) data.master_id (auto-generated if omitted)
description_field	(static value) 자동 발행: 임계 온도 초과로 긴급 점검 필요
default_priority	HIGH

Step 6 — Email notification (success branch)

Notify the responsible person when the work order is issued successfully.

1. Drag flow_send_email from the External category
2. Wire from the work order node's SUCCESS output
3. Inspector:

Option	Value
Display name	정비 담당자 이메일
to	ops@example.com
subject_template	[과열 정비] ${originator.id} 작업지시 ${data.work_order_id}
body_template	자산 ${originator.id} 의 온도가 ${data.value}°C 로 상승하여 자동으로 긴급 정비가 발행되었습니다.\n작업지시 ID: ${data.work_order_id}

Step 7 — Handle the failure branch

Issuing the work order itself may fail (for example, the asset was removed or there is a permission problem). Send an immediate push notification to the operator.

1. Drag flow_send_push from External
2. Wire from the work order node's FAILURE output
3. Inspector:

Option	Value
title	자동화 실패
body_template	${originator.id} 정비 자동 발행 실패: ${data.error}

Step 8 — Save and test run

1. Click the Save button at the top right. An automatic snapshot is stored so you can roll back later.
2. Click the Test Run button at the top right → enter the following in the JSON editor → Publish

{
  "type": "POST_TELEMETRY",
  "originator": { "entity_type": "Tag", "id": "MOTOR-001.TEMP" },
  "data":     { "value": 92.5 },
  "metadata": { "ts": 1746247200000, "tag_id": "MOTOR-001.TEMP", "asset_id": "MOTOR-001" }
}

✓ JSON OK (type=POST_TELEMETRY) appears at the bottom of the dialog → click Publish.

3. Check the following in the live debug panel:
   • Trigger node lights green → message passed
   • Filter node: since 92.5 > 80, it takes the TRUE branch
   • Transform node: originator changed to Asset/MOTOR-001
   • Work order node: SUCCESS, data.work_order_id assigned automatically
   • Email node: send attempted

Step 9 — Verify and deploy

1. Check on the work order screen that a new work order was registered
2. Check that the email arrived (test environment mailbox)
3. Test once more with a message below 80 °C (it should be blocked by the filter):

{ "type": "POST_TELEMETRY", "originator": {"entity_type":"Tag","id":"MOTOR-001.TEMP"},
  "data": {"value": 70}, "metadata": {"asset_id":"MOTOR-001"} }

The filter node branch is FALSE and the downstream nodes are gray — normal.

4. After verifying every branch, use Reset All Counts to reset the statistics window
5. Turn the Deploy toggle ON at the top

Now, in the real production environment, the moment a motor's temperature exceeds 80 °C a maintenance work order is issued automatically and an email goes to the responsible person.

Step 10 — Operational monitoring

It is a good idea to check the following for 5–10 minutes right after deployment.

Location	What to check
List screen	Whether the execution count for that flow row increases within a normal range (not runaway)
Live debug panel	Whether there are any node failures
Execution history screen	If there are failed messages, check the cause in NODE_ERROR
Received email	Whether notifications are being sent at an unintended frequency

Next steps

To evolve this flow:

• Add Retry wiring — back-off retry when email/push delivery fails (see Example 8)
• Automatic band adjustment — auto-correct the threshold to a 6-hour average ± 3σ (Example 6)
• Downtime accumulation — accumulate overheating history in an asset attribute (Example 14)
• Quality line isolation — automatically stop the line when overheating occurs repeatedly (Example 15)

---

Screen Layout

Flow consists of the following three screens.

Screen	Purpose
List	List, search, bulk deploy, and import registered flows
Edit	Place, connect, and configure nodes on the visual canvas
Execution history	View per-node execution logs and timelines

---

List Screen

Consists of the search/create area at the top and the flow listing table.

Top tools

Item	Description
Status filter	전체 / 배포 / 해제
Name/description search	Filters flows by keyword
New Flow	Creates an empty flow (enter name and description)
Import	Uploads a JSON obtained via export to restore a flow
Redeploy All	Reloads all enabled flows at once
Refresh	Reloads the list

Listing table

The list table is paginated 25 rows at a time, and each row also shows a processing-trend sparkline and an error-ratio donut chart. Chart state is preserved as you page through.

Column	Description
Select	Checkbox for bulk deploy/undeploy
Status	배포 / 해제 badge
Flow ID	Sequence ID in FLOW_NNNNN format
Name | Description	Metadata specified by the operator
Nodes	Number of nodes included
Triggers	Number of trigger nodes
Executions	Cumulative message processing count
Processing time	Node average / most recent processing time
Errors	Cumulative error count
Last modified	Time the graph was last saved
Actions	편집 / 삭제 buttons

Bulk deploy/undeploy

Deploys (enables) or undeploys (disables) the selected flows all at once. Only deployed flows receive trigger events.

Redeploy All

The Redeploy All button at the top reloads all enabled flows. Use it in the following situations.

• Right after bulk-importing graphs from outside
• When you want to re-register triggers that have their own scheduler, such as schedules, external MQTT subscriptions, or external DB polling
• When you suspect a cache consistency problem during operation

---

Edit Screen (Visual Canvas)

Drag nodes from the palette on the left of the canvas to place them, then click and drag from a node's output port to connect it to the next node.

Top toolbar

Button	Action
Name/description	Edit the flow metadata
Deploy/Undeploy	Immediately enable/disable the current flow
Export	Download the entire graph as a JSON file
Test Run	Inject an arbitrary JSON message, run once, and check the result (see How to Use Test Run below)
Save	Save the current graph to the server. An automatic snapshot is stored on save so you can roll back later

If you try to save a graph with no trigger, the warning "Can only be started by a manual test run" appears. You can intentionally omit triggers and use the flow for manual execution only.

How to use Test Run

Pressing the 테스트 실행 button opens a JSON editing dialog. Operators can write a message themselves and publish it once, so you can verify graph behavior without waiting for a trigger event.

Item	Description
Editor	JSON editor with line numbers and syntax highlighting. Write the message body freely
Validation indicator	If the type required field is present, ✓ JSON OK (type=X) is shown; if missing, ⚠ type 필수
Publish	The 발행 button injects the message into the dispatcher. Check the result in the live debug panel and execution history

Basic template example

{
  "type": "POST_TELEMETRY",
  "originator": { "entity_type": "Asset", "id": "MOTOR-001" },
  "data":     { "speed": 1500, "temp": 75.3 },
  "metadata": { "ts": 1746247200000, "site_id": "SITE-01" }
}

If you press Test Run while you have unsaved changes, a notice appears saying "The server runs the saved version." Save first if you want to verify your changes.

Left — Node palette

Categories can be collapsed and expanded, and the search input filters instantly.

Category	Color	Node count
Trigger	Gray	22
Filter	Blue	6
Transform	Green	8
Action — integration, storage, asset publishing, command invocation	Orange	7
Action — Domain CRUD	Orange	34
External	Purple	9
Control	Gray	7
Edge	Teal	22

Center — Canvas

Tool	Shortcut / operation	Action
Zoom in/out	Ctrl/⌘ + / Ctrl/⌘ - · mouse wheel	Canvas zoom
100%	Ctrl/⌘ 0	Reset zoom
Fit to screen	Ctrl/⌘ 1	Auto-fit so that all nodes are visible
Delete selected node	Del / Backspace	Delete the selected node/wire
Panning	Left-click an empty area and drag	Grab an empty canvas background without clicking a node and drag; the whole canvas moves with you

A minimap is displayed at the bottom right of the canvas; clicking the minimap jumps immediately to that position.

💡 Panning tip: dragging over a node moves the node — to move the canvas, always grab an empty background area with no nodes or wires. On large flows, panning is faster than the minimap.

Right — Node configuration

Clicking a node on the canvas shows that node's configuration form in the inspector on the right. Input fields are generated automatically based on the node type.

Input method	Description
Static value	Uses the value entered directly in the form as-is
*_field dynamic value	Extracts the value from a path in the message payload (for example data.tag_id, metadata.site_id). If no value exists, it falls back to the static value

Script nodes (filter, transform, switch) can be edited directly in a code editor inside the inspector, and can also be expanded into a separate dialog for editing on a larger screen.

The code editor font uses a readability-optimized monospace font stack (Cascadia Code · JetBrains Mono · Consolas · Menlo, in that order), and Korean comments also align reliably.

Resetting counts

Two reset buttons appear as icons at the top right of the node configuration panel. Hovering over each button shows a tooltip.

Icon button	Action
🩹 (bandage) — Reset error counts	Resets only the cumulative error counts of all nodes in this flow to 0
🔄 (circular arrows) — Reset all counts	Resets processing, error, and processing time counters plus flow-level statistics all to 0. Use it when you have finished operational verification and want to start statistics fresh

Both buttons take effect immediately without a confirmation dialog — only the statistics are reset; node behavior and message processing are unaffected.

Right — Live debug

The live debug panel is shown below the inspector.

Item	Description
Refresh interval	2 seconds
Level colors	INFO (blue) / WARN (yellow) / ERROR (red) shown on the left border
Displayed information	Node display name · processing time (ms) · message preview
Pause	Pauses refreshing via the toggle at the top right of the panel
Clear	Clears accumulated debug entries from the display only

A small indicator light appears at the top right of each node on the canvas.

Color	Meaning
Gray	Idle — has not received a message
Green	A message is passing through
Red	An error occurred during processing

The processing time is shown at the bottom right of the node in the form 평균 X · 최근 Y.

---

Message Structure

A message flowing through a flow consists of the following four areas.

{
  "type": "POST_TELEMETRY",
  "originator": {
    "entity_type": "Asset",
    "id": "MOTOR-001"
  },
  "data": { "speed": 1500, "temp": 75.3 },
  "metadata": {
    "ts": 1746247200000,
    "site_id": "SITE-01",
    "shift": "DAY",
    "tag_id": "MOTOR-001.SPEED"
  }
}

Area	Meaning
type	Message classification. The branching criterion for filter nodes
originator	The message's subject entity (which asset/tag/order it concerns)
data	The payload body
metadata	Context (time, site, shift, tag ID, and so on)

Message types

Type	Entry point
POST_TELEMETRY / TAG_POINT	Tag point ingestion
POST_ATTRIBUTES	Tag/asset metadata update
TAG_ALARM	Tag-level alarm
ENTITY_CREATED / UPDATED / DELETED	Entity lifecycle events
ASSET_DATA / ASSET_EVENT / ASSET_ALARM / ASSET_COMMAND / ASSET_AGGREGATION / ASSET_CONTEXT	Asset domain events
ASSET_HEALTH_STATUS / ASSET_CONNECTION_STATUS	Periodic asset evaluation (health/connection status)
OEE_EVENT / RAM_EVENT / EMS_EVENT	ISO analysis result events
OPC_STATUS / EDGE_STATUS	OPC/edge device status
DIAGNOSTIC / DOMAIN_CHANGED	Diagnostics / domain change
ALARM	Alarm occurrence
WEBHOOK	HTTP webhook received
KAFKA_INBOUND / MQTT_INBOUND	External topic received
TIMER	Schedule fired

Payload examples by trigger

When writing script nodes you need to know exactly which fields you can access. Below are actual message JSON examples produced by each trigger. All trigger messages commonly include type, originator, data, and metadata.

flow_on_tag_point — Tag point ingestion

Fires each time a value arrives for a single tag. This is the most common trigger.

{
  "type": "POST_TELEMETRY",
  "originator": { "entity_type": "Tag", "id": "MOTOR-001.SPEED" },
  "data": {
    "value": 1500.7,
    "quality": "GOOD",
    "ts": 1746247200123
  },
  "metadata": {
    "tag_id":      "MOTOR-001.SPEED",
    "site_id":    "SITE-01",
    "area_id":    "AREA-A",
    "line_id":    "LINE-1",
    "asset_id":   "MOTOR-001",
    "opc_id":     "OPC-LINE-1",
    "java_type":  "Float",
    "unit":       "rpm",
    "shift":      "DAY"
  }
}

Field	Meaning	Script access
data.value	Received value (numeric/string/boolean)	msg.data.value
data.quality	OPC quality (GOOD/BAD/UNCERTAIN)	msg.data.quality
data.ts	Reception time (epoch ms)	msg.data.ts
metadata.tag_id	Tag ID	msg.metadata.tag_id

flow_on_tag_alarm — Tag alarm occurrence

Fires the moment a tag's alarm band (hi/lo/...) is crossed.

{
  "type": "TAG_ALARM",
  "originator": { "entity_type": "Tag", "id": "MOTOR-001.TEMP" },
  "data": {
    "alarm_band": "HI_HI",
    "value":      95.3,
    "threshold":  90.0,
    "priority":   "ERROR",
    "band_message": "온도 위험"
  },
  "metadata": {
    "tag_id": "MOTOR-001.TEMP",
    "asset_id": "MOTOR-001",
    "site_id": "SITE-01",
    "ts": 1746247200123
  }
}

data.alarm_band value	Meaning
NORMAL / HI / LO / HI_HI / LO_LO / TRIP_HI / TRIP_LO	Numeric alarm levels
BOOL_TRUE / BOOL_FALSE	Boolean alarms

flow_on_asset_data / flow_on_asset_event — Asset events

Fires when an event aggregated at the asset level (CEP processing results, plugin evaluation results, and so on) occurs.

{
  "type": "ASSET_EVENT",
  "originator": { "entity_type": "Asset", "id": "MOTOR-001" },
  "data": {
    "event_type":  "STARTUP",
    "details":     { "rpm_target": 1500 }
  },
  "metadata": {
    "asset_id": "MOTOR-001",
    "site_id": "SITE-01",
    "ts": 1746247200123
  }
}

flow_on_asset_data carries asset-level time series data (data.values is a key-value map), while flow_on_asset_aggregation carries values aggregated by minute or hour.

flow_on_asset_health_status / flow_on_asset_connection_status — Periodic evaluation

The Platform evaluates asset-level health/connection status on a 1-minute cycle.

{
  "type": "ASSET_HEALTH_STATUS",
  "originator": { "entity_type": "Asset", "id": "MOTOR-001" },
  "data": {
    "status":       "WARN",
    "info_count":   12,
    "warn_count":   3,
    "error_count":  0,
    "prev_status":  "OK"
  },
  "metadata": { "asset_id": "MOTOR-001", "ts": 1746247200123 }
}

data.status value	Meaning
OK / WARN / ERROR / UNKNOWN	Health levels
CONNECTED / LATENT / ERROR / DISCONNECTED / UNKNOWN	Connection levels (connection_status)

A common pattern is to compare against prev_status and fire downstream actions only at the moment the state transitions.

flow_on_oee_event / flow_on_ram_event / flow_on_ems_event — Plugin events

Fires when OEE/RAM/EMS evaluation results per work order are updated.

{
  "type": "OEE_EVENT",
  "originator": { "entity_type": "WorkOrder", "id": "WO-20260512-001" },
  "data": {
    "oee":          0.78,
    "availability": 0.95,
    "performance":  0.85,
    "quality":      0.97,
    "good_count":   1560,
    "bad_count":    42,
    "target_count": 2000
  },
  "metadata": {
    "order_id":   "WO-20260512-001",
    "asset_id":   "LINE-1.PRESS",
    "shift_id":   "DAY-A",
    "ts": 1746247200123
  }
}

flow_on_opc_status / flow_on_edge_status — OPC/edge status

{
  "type": "OPC_STATUS",
  "originator": { "entity_type": "OPC", "id": "OPC-LINE-1" },
  "data": {
    "connection_status": "CONNECTED",
    "scan_status":       "START",
    "prev_status":       "DISCONNECTED"
  },
  "metadata": { "opc_id": "OPC-LINE-1", "edge_id": "EDGE-A", "ts": 1746247200123 }
}

flow_on_diagnostic — System diagnostic message

Fires when a diagnostic message generated by a server module arrives.

{
  "type": "DIAGNOSTIC",
  "originator": { "entity_type": "Module", "id": "cep-engine" },
  "data": {
    "level":   "WARN",
    "code":    "PATTERN_LAG",
    "summary": "EQL 패턴 평가 지연 1.2s",
    "module":  "cep-engine"
  },
  "metadata": { "ts": 1746247200123 }
}

data.level	Meaning
INFO / WARN / ERROR	Diagnostic severity

flow_on_domain_changed — Domain change events

Fires when domain entities such as assets, tags, sites, and work orders are created, updated, or deleted.

{
  "type": "DOMAIN_CHANGED",
  "originator": { "entity_type": "Asset", "id": "MOTOR-001" },
  "data": {
    "action":      "UPDATED",
    "before":      { "asset_name": "Motor1" },
    "after":       { "asset_name": "Motor 01 - Renamed" },
    "changed_by":  "admin"
  },
  "metadata": { "ts": 1746247200123 }
}

flow_on_entity_event — Entity lifecycle

Fires as three unified types: ENTITY_CREATED / ENTITY_UPDATED / ENTITY_DELETED. A single node receives all three lifecycle events.

{
  "type": "ENTITY_CREATED",
  "originator": { "entity_type": "Customer", "id": "CUST-9001" },
  "data": { "customer_name": "신규 고객", "external_id": "ERP-CUST-9001" },
  "metadata": { "ts": 1746247200123 }
}

flow_on_webhook — External HTTP push

A payload sent by an external system to POST /api/v4/flow/webhook/{flow_id} is converted directly into a message. Authenticated with {flow_id} in the URL path and the X-API-Key header.

{
  "type": "WEBHOOK",
  "originator": { "entity_type": "External", "id": "ERP" },
  "data": {
    "order_no":  "PO-20260512-001",
    "customer":  "ACME",
    "quantity":  1000
  },
  "metadata": {
    "http_method": "POST",
    "remote_addr": "10.20.0.55",
    "request_id":  "req-7c0a...",
    "ts": 1746247200123
  }
}

The entire JSON body sent by the external system is placed as-is into data. Only part of the HTTP headers (remote_addr/method/request_id) appears in metadata.

flow_on_mqtt_subscribe — MQTT topic subscription

{
  "type": "MQTT_INBOUND",
  "originator": { "entity_type": "Topic", "id": "factory/line1/events" },
  "data": { "event": "STARTUP", "rpm": 1500 },
  "metadata": {
    "topic":   "factory/line1/events",
    "qos":     1,
    "broker":  "tcp://mqtt.example.com:1883",
    "ts": 1746247200123
  }
}

flow_jdbc_poll — Periodic external DB polling

Executes the configured SELECT query periodically and fires one message per row.

{
  "type": "KAFKA_INBOUND",
  "originator": { "entity_type": "DB", "id": "mes_db" },
  "data": {
    "PO_NO":    "PO-20260512-001",
    "CUSTOMER": "ACME",
    "QTY":      1000,
    "DUE_DATE": "2026-05-20"
  },
  "metadata": {
    "datasource": "mes_db",
    "query":      "SELECT * FROM po WHERE status='NEW'",
    "row_index":  0,
    "ts": 1746247200123
  }
}

If there are 100 rows, 100 messages fire sequentially. To avoid processing the same row repeatedly, update a processing flag inside the SELECT query or add a processed_at column comparison condition.

flow_schedule — Time-based firing

Fires on a cron expression or a fixed interval. The payload is empty and only metadata.ts is filled in.

{
  "type": "TIMER",
  "originator": { "entity_type": "Schedule", "id": "daily-report" },
  "data": {},
  "metadata": {
    "cron":      "0 0 8 * * ?",
    "fired_at":  1746247200000,
    "ts":        1746247200000
  }
}

Cautions when transforming payloads

• originator.id is a domain ID — if you change it with the flow_change_originator node, the subsequent flow_save_attributes and flow_publish_asset_* action nodes operate against the new originator.
• When replacing data/metadata in a script, use direct assignment rather than a shallow copy (Object.assign) — modifying the original may affect other flows subscribing to the same trigger.
• All epoch time fields are in milliseconds (ms). If you need seconds, use Math.floor(msg.metadata.ts / 1000).

---

Node Catalog

Detailed node IDs and options can be checked in the inspector on the edit screen.

Triggers (22)

Every flow starts with a trigger node. There are no separate entry-point or terminal nodes.

Automatic domain reception — internal system events are dispatched automatically.

Category	Trigger nodes
Tag	flow_on_tag_point (telemetry), flow_on_tag_alarm
Asset	flow_on_asset_data, flow_on_asset_event, flow_on_asset_alarm, flow_on_asset_command, flow_on_asset_aggregation, flow_on_asset_context, flow_on_asset_health_status, flow_on_asset_connection_status
Plugin	flow_on_oee_event, flow_on_ram_event, flow_on_ems_event
OPC/Edge	flow_on_opc_status, flow_on_edge_status
Diagnostics/domain	flow_on_diagnostic, flow_on_domain_changed
Entity	flow_on_entity_event

There is no node called flow_on_alarm

Alarm triggers are split in two by target — tag alarms use flow_on_tag_alarm, and asset alarms use flow_on_asset_alarm. Examples in older documents used flow_on_alarm, but that name does not exist in the palette, so following them literally will leave you unable to find the node.

All trigger nodes support the *_pattern option (globs: *, ?) for per-message pre-filtering. Messages that do not match the pattern are not passed to downstream nodes and do not increment the execution count (treated as SKIPPED). To minimize operational load, we recommend filtering at the trigger stage first.

External entry

Node	Action
flow_on_webhook	Receives payloads pushed by external systems over HTTP
flow_on_mqtt_subscribe	Subscribes to a topic on an external MQTT broker
flow_jdbc_poll	Periodically reads SELECT results from an external database and fires per row

Time-based

Node	Action
flow_schedule	Cron/interval schedule — fires a TIMER message

Filters (6)

Node	Description
flow_msg_type_filter	type → TRUE if it is in the specified list
flow_originator_type_filter	originator.entity_type → TRUE if it is in the specified list
flow_script_filter	Evaluates a boolean with a script
flow_check_existence_field	Whether a specific field exists in data/metadata
flow_switch	Multi-case branching (a different relation per case)
flow_check_relation	Branches based on the relation from the previous step

Transforms (8)

Node	Display name	Description
flow_script_transform	Script transform	Transforms data/metadata with a script
flow_change_originator	Change originator	Changes originator to another entity
flow_rename_keys	Rename keys	Bulk-renames data field names
flow_template	Template	Generates ${path}-substituted text
flow_split	Split	Splits into one message per element if data is an array
flow_merge	Merge	Merges multiple messages over a time window — the opposite of flow_split
flow_flatten	Flatten	Lifts child keys of nested objects to the top level (data.data_map.x → data.x)
flow_to_email	Email transform	Converts a message into email format

Use flow_flatten when you want to reference a message that, like a tag point, has values nested one level deeper inside data_map directly as ${data.x} in later nodes.

Actions — Integration & Storage (2)

Node	Description
flow_save_tag_point	Stores tag points (same as the normal ingestion path)
flow_save_attributes	Partially updates tag/asset metadata

If you are looking for flow_dds_publish (free publishing to the internal message channel), it is not here but in the External Integration palette group.

Actions — Asset Event Publishing (4)

Publishes asset events through the same processing path as CEP (EQL) (consistent handling of persistent storage + cache + timeline + plugins + message channel).

Node	Channel
flow_publish_asset_event	Asset event
flow_publish_asset_context	Asset context
flow_publish_asset_aggregation	Asset aggregation
flow_publish_asset_command	Asset command

flow_asset_command_invoke — Execution, not publishing

Node	Display name	What it does
flow_asset_command_invoke	Invoke asset command	Synchronously executes a command defined on the asset and waits for the result

Easily confused with flow_publish_asset_command

• flow_publish_asset_command — publishes an asset command event to the channel. It publishes and is done.
• flow_asset_command_invoke — actually executes the command defined on the asset and waits for the result.

If you wanted the command to be executed but used the publish node, it will look as if nothing happened.

Actions — Domain CRUD (34)

All domain operations are delegated to the same domain services, so auditing and consistency are preserved. The *_field dynamic option lets you extract values from the message payload.

Domain	Create	Update	Delete
Asset	flow_create_asset	flow_update_asset	flow_delete_asset
Tag	flow_create_tag	flow_update_tag	flow_delete_tag
Site/Area/Line	flow_create_site	flow_update_site	flow_delete_site
WorkOrder	flow_create_work_order	flow_update_work_order	flow_delete_work_order
Alarm configuration (EQL)	flow_create_alarm_config	flow_update_alarm_config	flow_delete_alarm_config
Customer	flow_create_customer	flow_update_customer	flow_delete_customer
Product	flow_create_product	flow_update_product	flow_delete_product
Employee	flow_create_employee	flow_update_employee	flow_delete_employee
Calendar (shift)	flow_create_calendar	flow_update_calendar	flow_delete_calendar

Tag alarm band partial update (2)

Node	Description
flow_update_tag_alarm_band_numeric	Partially updates only the entered fields of a numeric alarm band (hi/lo/hi_hi/lo_lo/trip_hi/trip_lo/band_message/use_alarm)
flow_update_tag_alarm_band_boolean	Partially updates only the entered fields of a boolean alarm band (bool_true/bool_false/priority/message/use_alarm)

WorkOrder state transitions (5)

Because these call the domain service's state transition methods rather than updating numeric columns directly, OEE/RAM/EMS visibility is preserved.

Node	Transition
flow_start_work_order	WAIT → START
flow_pause_work_order	START → PAUSED
flow_resume_work_order	PAUSED → START
flow_end_work_order	START or PAUSED → END
flow_abort_work_order	START or PAUSED → ABORTED (abort_code, notes options)

Automatic NOT NULL enrichment — Create nodes automatically fill default values into NOT NULL columns. For example: work order status="WAIT" / master_id are MES master IDs (auto-backfilled if absent), customer manager information "admin"/"admin@example.com", employee org_id falls back to site_id, and insert_user_id="flow" on every row. In FK columns (for example customer_id/product_id), empty strings are converted to NULL.

Update nodes — partial update — the Customer/Product/Employee/Calendar and alarm band Update nodes first fetch the existing record, then merge and save only the fields you entered. Empty strings and null values are ignored, so existing values are preserved. If you need a full overwrite, use a Delete + Create combination.

A direct alarm trigger node is intentionally excluded. Alarms must be raised only through the flow_create_alarm_config path so that alarm history stays consistent.

Edge (22)

Calls the REST API of edge devices (OPC Agent) to automate OPC server registration, tag CRUD, tag value read/write, queries, and Docker app control. All nodes share semantically identical behavior, differing only in the default HTTP method (GET/POST/PUT/DELETE).

Group	Nodes
OPC server management	flow_edge_opc_create, flow_edge_opc_update, flow_edge_opc_delete, flow_edge_opc_start, flow_edge_opc_stop, flow_edge_opc_list
Tag management	flow_edge_tag_create, flow_edge_tag_update, flow_edge_tag_delete, flow_edge_tag_read, flow_edge_tag_write, flow_edge_tag_list
Queries	flow_edge_monitoring, flow_edge_info, flow_edge_transfer
Docker app control	flow_edge_app_list, flow_edge_app_inspect, flow_edge_app_start, flow_edge_app_stop, flow_edge_app_restart, flow_edge_app_logs, flow_edge_app_stats

The three query nodes

Node	Display name	What it does
flow_edge_monitoring	Monitoring	Retrieves edge monitoring indicators
flow_edge_info	Edge information	Retrieves edge ID, version, and uptime
flow_edge_transfer	Transport health	MQTT/Sparkplug transport status

Node	Display name	What it does
flow_edge_opc_list	OPC list	Retrieves the list of OPC servers on the edge
flow_edge_tag_list	Tag list	Retrieves the OPC tag list

The seven Docker app control nodes

Automates in a flow what you used to do by hand in the Docker panel of the Edge detail screen.

Node	Display name	What it does
flow_edge_app_list	App list	Lists edge Docker containers
flow_edge_app_inspect	App detail	Container detail (inspect)
flow_edge_app_start	Start app	Starts a container
flow_edge_app_stop	Stop app	Stops a container
flow_edge_app_restart	Restart app	Restarts a container
flow_edge_app_logs	App logs	Container logs (line=N)
flow_edge_app_stats	App stats	Container CPU / memory / I/O

App control nodes really do operate field equipment

app_stop and app_restart genuinely stop and restart containers running on the field edge. When you attach them to a trigger for automatic execution, keep the conditions narrow — attaching app_restart to a flapping alarm will restart the container over and over.

Common settings

Option	Description
url	Edge REST endpoint. Supports ${data.x}/${metadata.y} template substitution
method	HTTP method (if unset, the node's default — for example create=POST, update=PUT, delete=DELETE, read/monitoring=GET)
headers	JSON headers (for example {"Authorization":"Bearer ${TOKEN}"})
body_template	Request body (if unset, data is sent as-is; GET/DELETE send no body)
timeout_ms	Timeout (default 5000)

Response & branching

• data.response_status — HTTP status code
• data.response — response body (string)
• data.error — error message (on failure)
• SUCCESS (200–399) / FAILURE (otherwise, or on exception)

External Integration (9)

All external nodes support the *_field dynamic option.

Node	Dynamic options
flow_dds_publish	Free publishing to the internal message channel (not an external call, but it lives in this group)
flow_http_request	url_field / method_field / body_field
flow_kafka_publish	topic_field / key_field
flow_mqtt_publish	topic_field
flow_webhook_callback	url_field
flow_send_email	to_field / cc_field / subject_field / body_field
flow_send_sms	to_field / text_field
flow_send_push	title_field / body_field
flow_jdbc_query	SQL static (SELECT/INSERT/UPDATE/DELETE)

Flow Control (7)

Node	Description
flow_log	Debug log (level / prefix)
flow_noop	Pass through
flow_delay	Passes to the next node after delay_ms
flow_throttle	Limits max_msgs / window_ms (THROTTLED relation when exceeded)
flow_debounce	Fires only the last message after window_ms of stability
flow_merge	Accumulates input for window_ms and emits once as a data.merged array
flow_subflow	target_flow_id — calls another flow
flow_retry	max_attempts (default 3) / backoff_ms (default 1000) / backoff_multiplier (default 2.0). After the back-off wait, passes the message to the SUCCESS branch; on reaching the maximum attempts, the EXHAUSTED branch. metadata.retry_count / metadata.retry_exhausted are updated automatically

Recommended Retry wiring pattern

[risky_node] ─[FAILURE]─▶ [flow_retry] ─[SUCCESS]─▶ (다시 risky_node 로 루프 연결)
                                       └[EXHAUSTED]─▶ [에러 핸들러 / 알림]

---

Detailed Node Option Reference

Explains the options of complex nodes in detail — defaults, examples, and failure handling — rather than a one-line table.

flow_script_filter — Script filter

Option	Type	Default	Description
script	text	(required)	Evaluation expression — returns a boolean. true → TRUE branch / false → FALSE branch
script_type	enum	javascript	javascript / eql
on_error	enum	FALSE	On script exception — whether to send to the TRUE / FALSE / FAILURE branch

Script context

Variable	Meaning
msg.type	Message type (POST_TELEMETRY, etc.)
msg.data	Payload (modifiable, but meaningless in a filter)
msg.metadata	Context
msg.originator	The originator object

Examples

// 온도가 임계 초과 + 야간 시프트만
msg.data.value > 80 && msg.metadata.shift === 'NIGHT'

// 사이트별 임계 분기
var th = {'SITE-A': 80, 'SITE-B': 90, 'SITE-C': 75};
msg.data.value > (th[msg.metadata.site_id] || 100);

flow_script_transform — Script transform

Option	Type	Default	Description
script	text	(required)	Transformation expression — modify the msg object or return a new one
script_type	enum	javascript	javascript / eql
mode	enum	mutate	mutate (in-place) / return (use the returned value)

Examples

// data 에 계산 필드 추가
msg.data.fahrenheit = msg.data.value * 9/5 + 32;
msg.metadata.processed_at = Date.now();

// 페이로드 통째로 교체 (mode=return)
return {
  type: 'WEBHOOK',
  originator: msg.originator,
  data: { temp: msg.data.value, level: msg.data.value > 80 ? 'HIGH' : 'OK' },
  metadata: msg.metadata
};

In mutate mode, a return statement is ignored even if present. To replace with a new object, you must set mode=return.

flow_switch — Multi-branch

Option	Type	Default	Description
cases	array	(required)	Array of [{expression, relation}] — evaluated top to bottom; the first match wins
default_relation	string	DEFAULT	When no case matches
script_type	enum	javascript	—

Example

// cases 설정
[
  { "expression": "msg.data.value > 90", "relation": "CRITICAL" },
  { "expression": "msg.data.value > 80", "relation": "WARN" },
  { "expression": "msg.data.value > 70", "relation": "INFO" }
]
// default_relation: "NORMAL"

Downstream nodes can then handle the four branch labels (CRITICAL/WARN/INFO/NORMAL) differently.

flow_retry — Automatic retry

Automatically recovers from transient failures of external IO nodes.

Option	Type	Default	Description
max_attempts	int	3	Maximum number of attempts (exceeding this yields EXHAUSTED)
backoff_ms	long	1000	Initial wait time (ms)
backoff_multiplier	double	2.0	Exponential back-off multiplier — 1st 1s → 2nd 2s → 3rd 4s
max_backoff_ms	long	30000	Upper bound for a single wait
jitter_pct	int	0	Random ±N% jitter on the back-off (avoids thundering herd)

metadata automatic enrichment

Field	Meaning
metadata.retry_count	Number of attempts so far
metadata.retry_exhausted	If true, enters the EXHAUSTED branch
metadata.retry_last_error	Reason for the last failure

If the total back-off exceeds 60 seconds, the trigger processing queue may back up. If the external system is consistently slow, limit the intake rate first with flow_throttle.

flow_on_webhook — HTTP trigger

Option	Type	Default	Description
auth_required	boolean	true	Whether the X-API-Key header is required — if off, anyone can call it
allowed_origins	csv	*	CORS Origin whitelist
max_body_kb	int	256	Body size limit (exceeding it returns 413)
payload_pattern	glob	*	Message pre-filtering glob

How to call

curl -X POST \
  https://platform.example.com/api/v4/flow/webhook/{flow_id} \
  -H "X-API-Key: {edge_or_token_key}" \
  -H "Content-Type: application/json" \
  -d '{"order_no":"PO-001","customer":"ACME","quantity":1000}'

• {flow_id} in the path can be copied from the list screen
• X-API-Key is a token issued on the edge API Key or API authentication token screen
• Responses: 200 OK (enqueued into the message queue) / 401 (authentication failure) / 404 (flow missing or not deployed) / 413 (body too large)

flow_http_request — External HTTP call

Option	Type	Default	Description
url	string	(required)	URL to call. ${data.x} template substitution
url_field	string	—	When taking the URL dynamically from the payload — for example data.endpoint
method	enum	GET	GET / POST / PUT / DELETE / PATCH
method_field	string	—	Take the method dynamically from the payload
headers	json	{}	In {"Authorization": "Bearer ${TOKEN}"} form
body	text	—	Static body (supports template substitution)
body_field	string	—	When taking the body from the payload — usually data
timeout_ms	int	5000	Upper bound for waiting on a response
follow_redirect	boolean	true	Automatically follow 3xx redirects
verify_ssl	boolean	true	TLS certificate verification (turn off for testing only)

Response payload enrichment

Field	Meaning
data.response_status	HTTP status code (200 / 404 / 500 ...)
data.response_body	Response body (auto-parsed if JSON)
data.response_headers	Response header object

Branching

• SUCCESS — 2xx/3xx
• FAILURE — 4xx/5xx, or exception/timeout

flow_send_email — Send email

Option	Type	Default	Description
to	string	—	Static recipients (comma-separated)
to_field	string	—	Extract recipients from the payload — for example data.recipient
cc / cc_field	string	—	CC
bcc / bcc_field	string	—	BCC
subject / subject_field	string	(one required)	Subject — template substitution
body / body_field	text	(one required)	Body (HTML allowed)
is_html	boolean	true	Turn off for plain-text mail
attachments	json	[]	[{"url":"...","filename":"..."}]

SMTP settings must be registered in advance by the operator under mail settings in System → Settings. Until they are registered, all email nodes fall into the FAILURE branch.

flow_jdbc_poll — External DB polling

Option	Type	Default	Description
datasource_id	string	(required)	Identifier of an external DB registered in System → Settings
query	sql	(required)	SELECT query — returns at most 1,000 rows at a time
poll_interval_ms	int	60000	Polling interval (default 1 minute)
marker_column	string	—	"Last processed time" column — SELECTs only rows after the marker
marker_initial	string	1970-01-01 00:00:00	Marker start value on the first poll
row_limit	int	1000	Maximum rows per poll (safe even if exceeded)
on_error_continue	boolean	true	On DB error, record diagnostics only and continue with the next poll

marker_column usage example

SELECT po_no, customer, qty, created_at
FROM po
WHERE created_at > :marker
ORDER BY created_at

→ The maximum created_at value received in the last poll is automatically substituted where :marker appears.

flow_kafka_publish / flow_mqtt_publish — External publishing

Option	Type	Default	Description
broker	string	(required)	kafka:9092 or tcp://mqtt:1883
topic	string	—	Static topic — ${data.x} substitution possible
topic_field	string	—	Extract the topic from the payload (for example data.target_topic)
key / key_field	string	—	(Kafka only) message key
body / body_field	json/text	(one required)	Body to publish — if unset, data as-is
qos	int	1	(MQTT only) 0/1/2
retain	boolean	false	(MQTT only) Retained flag

flow_publish_asset_* — Publish asset event

Asset events (event/context/aggregation/command) are published to a unified channel recognized simultaneously by CEP, plugins, and the timeline. Common options for the four nodes:

Option	Type	Default	Description
asset_id	string	—	Static asset ID
asset_id_field	string	—	Extract the asset ID from the payload (usually metadata.asset_id)
event_type	string	—	Asset event classification (for example STARTUP, SHUTDOWN, MAINTENANCE)
event_type_field	string	—	Extract the classification from the payload
payload	json	${data}	Body to publish — if unset, data as-is

In the case of flow_publish_asset_command, it is delivered immediately to the asset's command reception topic (asset_id/cmd/{event_type}) and reaches the edge.

flow_create_* / flow_update_* — Domain CRUD

All Create/Update nodes share the following option pattern.

Option	Type	Description
{컬럼명}	string	Static value (NULL/default if not entered)
{컬럼명}_field	string	Extract the value from the payload (data.foo / metadata.bar)
id_strategy	enum	auto (system-issued) / field (extracted from {도메인}_id_field)
on_duplicate	enum	error (default) / skip / update — Create nodes only

Automatic NOT NULL enrichment

Create nodes automatically fill default values into NOT NULL columns.

Domain	Auto-filled columns
Work order	status="WAIT" / master_id (auto-backfilled if absent) / insert_user_id="flow"
Customer	Manager information "admin" / "admin@example.com" (if not registered)
Employee	org_id=site_id (fallback)
Common	insert_date=now() / insert_user_id="flow"

Handling of empty strings in FK columns

If an empty string "" arrives in an FK column (customer_id/product_id, etc.), it is automatically converted to NULL. In JS, msg.data.customer_id = '' is safer than delete msg.data.customer_id.

flow_edge_* — Edge REST call

The 22 nodes that call edge device REST endpoints share common options.

Option	Type	Default	Description
edge_id	string	—	Static edge ID
edge_id_field	string	—	Extract the edge ID from the payload
path	string	(per-node default)	Edge REST path (for example /api/v1/opc, /api/v1/app/grafana/start)
body_template	text	—	Request body — if unset, data as-is
timeout_ms	int	5000	—

Automatic authentication

When called, edge nodes automatically look up that edge's api_key from the mm_edge master table and attach it as the X-API-Key header. No separate operator configuration is needed.

Response payload

Field	Meaning
data.edge_response_status	Edge response code
data.edge_response	Response body
data.edge_id	The target edge ID (for confirmation)

Branching is identical to flow_http_request (SUCCESS / FAILURE).

---

Writing Script Nodes

The flow_script_filter / flow_script_transform / flow_switch nodes support two expression languages.

JavaScript (recommended default)

Standard ECMAScript syntax. Multi-line code, var/let/const, functions, and object literals are all supported.

Bindings

Variable	Description
msg	The whole message. msg.data.x, msg.metadata.topic, msg.type, and msg.originator.id are all directly accessible
data	Shorthand alias for msg.data
metadata	Shorthand alias for msg.metadata

Filter example

data.temp > 80

Transform example

data.temp_f = data.temp * 1.8 + 32;
data.alert = data.temp > 80 ? 'HIGH' : 'OK';
msg

Switch example (boolean per case)

data.t > 100   // case "Critical"

Frequently used transformation patterns

// 1) 단위 변환 (섭씨 → 화씨) + 라벨링
data.temp_f = data.temp * 1.8 + 32;
data.alert  = data.temp > 80 ? 'HIGH' : 'OK';
msg

// 2) 메타데이터 보강 — 시간대·시프트 자동 부여
const h = new Date(metadata.ts).getHours();
metadata.shift = (h >= 6 && h < 18) ? 'DAY' : 'NIGHT';
msg

// 3) 외부 페이로드를 도메인 모델로 매핑 (MES PO → WorkOrder)
const po = data;
data = {
  master_id: 'WO-MES-' + po.po_no,
  asset_id:  po.line_id || 'UNASSIGNED',
  title:     po.product_name + ' (' + po.qty + ')',
  due_date:  po.delivery_date,
  qty:       po.qty
};
msg

// 4) 실패 분기로 명시적 라우팅 (필수 필드 누락 시)
if (!data.tag_id || data.value == null) throw new Error('필수 필드 누락');
msg

// 5) 배열 분할 후 데이터 정제 — split 노드 후에 사용
data.value     = parseFloat(data.raw);
data.threshold = data.value > 100;
msg

Frequently used filter patterns

// 우선순위 화이트리스트
['ERROR', 'CRITICAL'].includes(data.priority)

// 시간대 기반 필터 (주간만 허용)
new Date(metadata.ts).getHours() >= 8 && new Date(metadata.ts).getHours() < 20

// 자산 ID 패턴 매칭
/^MOTOR-.*$/.test(originator.id)

// 임계값 + 안정성 (값이 5번 이상 누적된 경우)
data.value > data.threshold && data.consecutive_count >= 5

User scripts run in a secure sandbox; file, network, thread, and arbitrary class access are all blocked. If you need to call an external system, wire in a separate external integration node such as flow_http_request.

EQL expressions

For compatibility with existing EQL users. Only a single expression is supported (multi-line code and semicolon separation are not), and only side-effect patterns can be used.

#msg.getData().getInt('temp') > 80

If you need multiple operations, use JavaScript.

JavaScript runtime specification

Scripts run in an isolated sandbox. You need to know exactly what is and is not possible in order to write stable scripts.

Available (✅)

Feature	Notes
Standard ECMAScript	var/let/const, functions, classes, destructuring, spread, ?., ??, etc.
Object literals	{ key: value, ... }
Array methods	map / filter / reduce / forEach / find / some / every / flat / slice
String methods	split / replace / includes / match / padStart / repeat
Math functions	All of Math.*
JSON	JSON.parse / JSON.stringify (but if data is already an object, no need to stringify again)
Date	new Date() / Date.now() / getHours() / toISOString(), etc.
Regular expressions	/pattern/ literals + the RegExp constructor
Throwing errors	throw new Error('...') — automatically branches to FAILURE
try/catch/finally	Exception handling

Not available (❌)

Feature	Reason / alternative
Network calls (fetch / XMLHttpRequest)	Blocked by the sandbox — wire in a separate flow_http_request node
File system (require('fs'))	Blocked by the sandbox
Threads (setTimeout / setInterval / Worker)	Synchronous execution only — use the flow_delay node if you need a delay
require / import	External modules cannot be loaded — define needed functions in the same script
eval / new Function(string)	Blocked for security
Arbitrary Java classes	Not exposed to user code, even in EQL compatibility mode
process / global / window	Not defined
WebSocket / EventSource	Blocked — receive messages via trigger nodes

Execution limits — there are none

Scripts have no time or memory limits

Older documents included a table stating a 500 ms execution time, 16 MB memory, stack 1024, and 2 MB output, but none of the four are enforced.

• It may look as if you can use timeout_ms in the node settings, but script nodes do not read that value. (ScriptTransformNode and ScriptFilterNode read only script and language.)
• No time, statement count, or memory limit is configured on the JS execution context either.

As a result, a script like while (true) {} occupies a worker thread indefinitely. The overall flow limit of 30 seconds (flow_timeout_ms) is checked when the execution loop picks up the next task, so an infinite loop inside a script never reaches that check.

Write only scripts that terminate on their own. Avoid infinite loops, huge iterations, and accumulating large strings, and always put an upper bound on loops.

What actually is blocked — the security sandbox

Instead of time and memory, anything that reaches outside is firmly blocked.

Item	Status
Java class access (Java.type, etc.)	Blocked
File and network IO	Blocked
Thread creation	Blocked
Native access	Blocked
Experimental options	Blocked
Host object access	Only what is on the allow list

If you need an external call, do not make it inside the script — separate it into a dedicated node such as flow_http_request. It will not work in a script anyway, and dedicated nodes do actually enforce timeout_ms.

Do not cram heavy processing into a single script; split it across multiple nodes.

msg return convention

// flow_script_transform 의 두 가지 모드
// 1) mutate 모드 (기본) — msg 객체를 직접 수정, 마지막에 msg 또는 아무 값 반환
data.foo = 'bar';
msg

// 2) return 모드 — 완전히 새 객체로 교체
return {
  type: msg.type,
  originator: msg.originator,
  data: { ...data, foo: 'bar' },
  metadata: msg.metadata
};

Standard time and locale

Item	Behavior
Server system time zone	UTC (using epoch ms directly is recommended)
Displaying Korean time	toLocaleString('ko-KR', { timeZone: 'Asia/Seoul' })
Computing Korean time	new Date(ts + 9*3600000).getUTCHours() or the locale above
Timestamps at whole seconds	Math.floor(Date.now() / 1000) * 1000

---

Memory-Safe Coding Patterns

Script nodes have a 16 MB memory limit, but if small memory leaks accumulate in frequently invoked nodes, the engine's GC burden grows. Avoid the following patterns.

Anti-pattern 1 — A closure holding large data

// ❌ 나쁜 예 — 큰 배열을 변환 함수에 캡쳐
const heavy = data.records || [];          // 1만 건
const summarize = (item) => heavy.find(r => r.id === item.id);
data.matched = data.targets.map(summarize);
msg

// ✅ 좋은 예 — 인덱스 미리 만들고 함수 안에서만 사용
const index = {};
(data.records || []).forEach(r => { index[r.id] = r; });
data.matched = data.targets.map(t => index[t.id]);
data.records = undefined;     // 변환 후 큰 원본 제거
msg

Anti-pattern 2 — Passing a huge array through as-is

// ❌ data.records 가 10,000 건이면 모든 후속 노드에서 메모리 차지
msg

// ✅ 필요한 통계만 남기고 원본 제거
data.summary = {
  count: (data.records || []).length,
  total: (data.records || []).reduce((s, r) => s + r.value, 0)
};
delete data.records;
msg

Anti-pattern 3 — Deep object copies

// ❌ JSON.parse(JSON.stringify(obj)) 는 큰 객체에서 매우 느림
data.copy = JSON.parse(JSON.stringify(data.original));

// ✅ 얕은 복사 또는 필요한 필드만 직접 선택
data.summary = { id: data.original.id, name: data.original.name };

Anti-pattern 4 — Regex explosion (catastrophic backtracking)

// ❌ (a+)+b 형태의 중첩 그룹은 입력에 따라 지수 시간
const re = /^(a+)+b$/;
if (re.test(data.text)) ...

// ✅ 비포획 그룹 + atomic 그룹 패턴 또는 단순 매칭
const re = /^a+b$/;
if (re.test(data.text)) ...

Anti-pattern 5 — let accumulator variables outside the function

// ❌ 함수 밖 let 은 매 노드 호출마다 0으로 리셋되지만, 의도와 다르게 동작 가능
let counter = 0;
data.items.forEach(() => counter++);
data.counter = counter;

// ✅ 명시적으로 함수 안에서 선언
data.counter = data.items.length;     // 같은 결과, 더 명확

Anti-pattern 6 — Nested try/catch with unbounded retries

// ❌ 실패해도 다시 던지지 않으면 그래프가 잘못된 분기로 이동
try {
  riskyCall();
} catch (e) { /* 무시 */ }
msg

// ✅ 실패 의도면 throw, 정상 의도면 명시적으로 기록
try {
  riskyCall();
  data.status = 'OK';
} catch (e) {
  data.status = 'ERROR';
  data.error_msg = e.message;
}
msg

An exception you throw inside a script goes to the FAILURE branch and the message is preserved. External IO nodes have their own branches, so do not wrap them in try/catch.

Diagnosing memory pressure

There is no code such as NODE_SCRIPT_OOM

Older documents presented NODE_SCRIPT_OOM, FLOW_MSG_TRIMMED, and ENGINE_GC_PRESSURE as signs of memory pressure, but the product does not produce such code. There is no script memory limit and no automatic payload truncation.

Judge memory pressure from metrics and symptoms, not from code.

What to watch	Pressure signal
FLOW_EXECUTOR_DROPPED_COUNT	Starts rising from 0 — the processing pool is saturated and messages are being dropped
FLOW_EXECUTOR_QUEUE_SIZE	Keeps increasing instead of shrinking
FLOW_EXECUTION_TIME	Peak values spike to several times normal
Server logs	work_queue full ... task dropped warnings, FlowExecutor timeout warnings
JVM	Increasing GC time, rising heap usage

If you see these signals, start by inspecting flows that handle large payloads and check whether scripts are accumulating large strings or arrays.

---

Mobile Push Notification Channel

Use the flow_send_push node to send push notifications to the operators' mobile app.

Node options

Option	Type	Default	Description
to / to_field	string	(one required)	Recipient user IDs (comma-separated) or a payload path
title / title_field	string	(one required)	Notification title (50 characters or fewer recommended)
body / body_field	text	(one required)	Body (120 characters or fewer recommended)
priority	enum	NORMAL	NORMAL / HIGH — HIGH is shown even on the lock screen
sound	enum	default	default / silent / a custom sound
data	json	{}	Additional data for the app to handle (4 KB payload limit)
deep_link	string	—	Screen to open when the notification is tapped (for example pp://asset/MOTOR-001)
ttl_sec	int	86400	Retention time in seconds if undelivered — automatically deleted on expiry

Automatic per-user token routing

When an operator first logs in to the mobile app, the device token is automatically registered under Security → API authentication tokens. In flows you do not handle tokens directly — just specify the user ID.

• If both iOS and Android tokens are registered, both devices receive the notification
• If a token becomes invalid (app deleted, etc.), it is automatically deregistered

Simple send example

[flow_on_asset_alarm]
  ↓ where priority='ERROR'
[flow_send_push]
   to_field:    "metadata.responsible_user"
   title:       "🚨 ${metadata.asset_id} 알람"
   body_field:  "data.band_message"
   priority:    HIGH
   deep_link:   "pp://alarm/view/${data.alarm_id}"

Multilingual push

// flow_script_transform — 사용자 로케일에 따라 본문 분기
const locale = metadata.user_locale || 'ko-KR';
const templates = {
  'ko-KR': { title: '🚨 ${asset} 위험 알람',           body: '값 ${value}, 즉시 점검 바랍니다.' },
  'en-US': { title: '🚨 ${asset} Critical Alarm',     body: 'Value ${value}, please inspect immediately.' },
  'ja-JP': { title: '🚨 ${asset} 危険警報',            body: '値 ${value}, 即時点検が必要です。' }
};
const tpl = templates[locale] || templates['ko-KR'];
data.push_title = tpl.title.replace('${asset}', metadata.asset_id).replace('${value}', data.value);
data.push_body  = tpl.body.replace('${value}', data.value);
msg

Then use title_field=data.push_title / body_field=data.push_body of flow_send_push.

Grouped notifications (inbox style)

Bundle several alarms into a single notification (per 5 minutes):

[flow_on_asset_alarm]
   ↓
[flow_merge window=300s]
   ↓ (data.merged 배열)
[flow_script_transform — 요약 만들기]
   ↓ data.push_body="알람 N건: A자산, B자산, ..."
[flow_send_push]

Automatic escalation when the user is unavailable

Escalate automatically to SMS or email 30 minutes after an unacknowledged push.

[flow_on_asset_alarm priority=ERROR]
   ↓
[flow_send_push]
   ↓ SUCCESS
   ↓ data.alert_id = response_id
[flow_delay 1800s (30분)]
   ↓
[flow_http_request GET /api/alert/${alert_id}/status]
   ↓ (response_body.read=false 면)
[flow_script_filter (data.response_body.read === false)]
   ↓ TRUE
[flow_send_sms]
   to_field: "metadata.responsible_phone"
   text:     "푸시 미확인 30분 경과: ${data.title}"

Recommended send frequency limits

Priority	Recommended frequency
HIGH (lock screen display)	5 or fewer per user per hour — prevents alarm fatigue
NORMAL	20 or fewer per user per hour

Wire a flow_throttle node before sending, or stabilize alarms from the same asset with flow_debounce before sending.

---

Automatic External Auth Token Refresh

A pattern for automatically refreshing expiring tokens such as OAuth2 when calling external systems.

Simple refresh — time-based (cron)

[flow_schedule cron='0 */50 * * * ?']    ← 50분마다 (만료 1시간 전)
   ↓
[flow_http_request]
   url:    "https://auth.example.com/oauth2/token"
   method: POST
   body:   "grant_type=client_credentials&client_id=${creds.client_id}&client_secret=${creds.client_secret}"
   ↓
[flow_script_transform]
   ↓ data.access_token = data.response_body.access_token
[flow_save_attributes]    ← 자격 증명 저장소에 저장
   target_id:   "creds:erp_api"
   attributes:  {"access_token": "${data.access_token}", "expires_at": ${data.response_body.expires_in * 1000 + ts}}

Then, in another flow's flow_http_request:

Headers: Authorization: Bearer ${creds:erp_api.access_token}

Proactive refresh — when a 401 is received

[main flow]
   ↓
[flow_http_request]
   ↓ SUCCESS → 정상 처리
   ↓ FAILURE
[flow_script_filter (response_status === 401)]
   ↓ TRUE
[flow_subflow target_flow_id="refresh-token"]    ← 토큰 갱신
   ↓
[원래 노드로 루프]    ← 갱신된 토큰으로 재시도

Using a refresh token

[flow_http_request]
   url:    "https://auth.example.com/oauth2/token"
   method: POST
   body:   "grant_type=refresh_token&refresh_token=${creds.refresh_token}"
   ↓
[flow_script_transform]
   // 새 access_token + 새 refresh_token (rotation)
   data.access_token  = data.response_body.access_token;
   data.refresh_token = data.response_body.refresh_token;
   data.expires_at    = Date.now() + data.response_body.expires_in * 1000;
   msg
   ↓
[flow_save_attributes]

Automatic alarm on token expiry threshold

[flow_schedule cron='0 0 * * * ?']    ← 매시
   ↓
[flow_jdbc_query]
   sql: "SELECT id, expires_at FROM credentials WHERE expires_at < NOW() + INTERVAL '1 day'"
   ↓
[flow_split]
   ↓
[flow_send_email]
   subject: "API 토큰 만료 임박: ${data.id}"
   body:    "${data.id} 토큰이 ${data.expires_at} 만료 예정입니다."

Recommended locations for storing credentials

Kind	Recommended location
Static (rarely changes)	The credential store under System → Settings
Dynamic (auto-refreshed)	Store in asset metadata using flow_save_attributes with the pattern above
Per-user OAuth	The Security → API authentication tokens screen

Do not leave credentials in plaintext inside the graph — always write them as external store references (${creds.x}). That way, plaintext tokens are not included in the JSON during export/import either.

---

Use the language option to choose which language to use (JS or EQL).

---

Execution History Screen

Lets you view per-node execution events in chronological order.

Search conditions

Item	Description
Time	Specifies the query period
Level	전체 / INFO / WARN / ERROR
Flow ID	Filters to a specific flow
Message ID	For tracing a single message
Count	Most recent 200/500/1,000 entries

Quick time ranges

Jump instantly using the buttons at the top of the timeline panel: 10분전 / 30분전 / 1시간전 / 6시간전 / 12시간전 / 전체기간.

Result columns

Column	Description
Level	INFO / WARN / ERROR
Time	Time the event occurred
Event	FLOW_START / NODE_IN / NODE_OUT / NODE_ERROR / FLOW_END
Flow ID	Which flow it was
Node	Node display name
Node type	For example flow_script_transform
Relation	SUCCESS / FAILURE / TRUE / FALSE / MATCH / NO_MATCH / DEFAULT / THROTTLED / EXHAUSTED (all uppercase)
Message	Summary of message type, subject, relation, processing time, and data preview

You can export the current query results with the CSV Download button.

Execution history is retained for 7 days. If you need long-term retention, ship it to an external log system.

---

General Workflow

1. Create a new flow — list screen 새 플로우 button → enter name and description
2. Enter the edit screen — an empty canvas opens automatically
3. Place a trigger node — drag a trigger node from the left palette
4. Add processing nodes — place filter → transform → action in order and connect them with wires
5. Configure nodes — click each node and enter its options in the inspector on the right
6. Save — the 저장 button at the top right (an automatic snapshot is stored)
7. Test run — inject an arbitrary message and check the result
8. Deploy — enable with the 배포 toggle → runs automatically when trigger events arrive
9. Monitor — check via the live debug panel and the execution history screen

---

Example Flows

Each example consists of a node wiring diagram + key node settings + a description of the behavior. The JSON-form node settings correspond 1:1 to the values you enter in the inspector on the right.

Example 1: Automatic MES work order creation

Polls MES every minute for new POs and converts them into work orders.

[flow_schedule: 매분]
   │
   ▼ TIMER
[flow_http_request: MES /api/po/list?status=NEW]
   │
   ▼ SUCCESS
[flow_split: data → 각 PO별 메시지]
   │
   ▼
[flow_script_transform: PO → WorkOrder 매핑]
   │
   ▼
[flow_create_work_order]
   │
   ├── SUCCESS → [flow_log: 워크오더 생성됨]
   └── FAILURE → [flow_send_email: 실패 알림]

Node settings

Node	Key settings
flow_schedule	cron: 0 * * * * ? (every minute at second 0)
flow_http_request	method: GET, url: https://mes.example.com/api/po/list?status=NEW, headers: {"Authorization":"Bearer ${MES_TOKEN}"}
flow_split	path: data (split as an array)
flow_script_transform	language: JS, see the script below
flow_create_work_order	master_id_field: data.master_id, asset_id_field: data.asset_id, title_field: data.title
flow_send_email	to_field: metadata.alert_to, subject: [MES 동기화 실패] ${data.po}

Script Transform example

data.master_id = 'WO-MES-' + data.po;
data.title     = data.product_name + ' (' + data.qty + ')';
data.asset_id  = data.line_id || 'UNASSIGNED';
data.due_date  = data.delivery_date;
metadata.alert_to = 'ops@example.com';
msg

Example 2: Publishing an asset command from an external webhook

Validates an HTTP payload sent by an external system and converts it into an internal asset command.

[flow_on_webhook]
   │
   ▼ WEBHOOK
[flow_script_filter: payload 검증]
   │
   ├── TRUE
   ▼
[flow_publish_asset_command]
   │
   ├── SUCCESS → [flow_log]
   └── FAILURE → [flow_webhook_callback: 외부에 실패 통보]

How to call: sending a JSON body with POST /flow/webhook/{flow_id} fires the trigger with that body placed as-is in the data area.

Script Filter example

// 인증 토큰 일치 + 필수 필드 존재 검사
if (data.token !== 'EXPECTED_TOKEN') return false;
if (!data.asset_id || !data.cmd_key) return false;
true

Example 3: Alarm → automatic emergency work order

When an alarm of ERROR grade or higher is received, issues an emergency maintenance work order against the parent asset.

[flow_on_tag_alarm]
   │
   ▼ ALARM
[flow_script_filter: priority = ERROR/CRITICAL]
   │
   ├── TRUE
   ▼
[flow_change_originator: Tag → 상위 Asset]
   │
   ▼
[flow_create_work_order: 긴급 정비]
   │
   └── SUCCESS → [flow_send_email: 정비 담당자]

Script Filter example

['ERROR', 'CRITICAL'].includes(data.priority)

Example 4: External DB synchronization — bulk asset registration

Polls a legacy DB for new equipment rows and registers them automatically as assets.

[flow_jdbc_poll: SELECT * FROM legacy_assets WHERE sync_status='NEW']
   │
   ▼
[flow_script_transform: 컬럼 매핑]
   │
   ▼
[flow_create_asset]
   │
   ├── SUCCESS → [flow_jdbc_query: UPDATE legacy_assets SET sync_status='OK' WHERE id=?]
   └── FAILURE → [flow_log: ERROR + flow_send_push]

Node settings

Node	Key settings
flow_jdbc_poll	dsn: external DB connection, sql: SELECT * FROM legacy_assets WHERE sync_status='NEW' LIMIT 100, interval_ms: 60000
flow_create_asset	asset_id_field: data.legacy_id, asset_name_field: data.name, site_id_field: data.plant_code
flow_jdbc_query	sql: UPDATE legacy_assets SET sync_status='OK' WHERE id=?, params_field: data.legacy_id

Example 5: Automatic work order state transitions

Automatically transitions work order states based on run/stop events from assets.

[flow_on_asset_event]
   │
   ▼ ASSET_EVENT
[flow_switch: data.event_type 기준]
   │
   ├── case "RUN"   ─▶ [flow_start_work_order] ─▶ [flow_log]
   ├── case "STOP"  ─▶ [flow_pause_work_order] ─▶ [flow_log]
   ├── case "DONE"  ─▶ [flow_end_work_order]   ─▶ [flow_log]
   └── DEFAULT      ─▶ [flow_noop]

Switch node case examples

• RUN: data.event_type === 'RUN'
• STOP: data.event_type === 'STOP'
• DONE: data.event_type === 'DONE' && data.qty_done >= data.qty_planned

When a state transition fails (invalid current state), it is automatically routed to FAILURE, so you can wire this safely without a separate Filter.

Example 6: Automatic alarm band adjustment

Dynamically adjusts a tag's upper/lower alarm bands based on asset aggregation (for example a 6-hour average).

[flow_on_asset_aggregation]
   │
   ▼ ASSET_AGGREGATION
[flow_script_transform: 통계 → 임계값 산출]
   │
   ▼
[flow_update_tag_alarm_band_numeric]
   │
   ├── SUCCESS → [flow_log]
   └── FAILURE → [flow_send_email: 운영자 알림]

Script Transform example

// 6h 평균 ± 3σ 를 임계값으로 사용
const mean   = data.mean;
const stddev = data.stddev || 1;
data.tag_id  = originator.id + '.TEMP';
data.hi      = mean + 3 * stddev;
data.lo      = mean - 3 * stddev;
data.hi_hi   = mean + 4 * stddev;
data.lo_lo   = mean - 4 * stddev;
data.use_alarm = true;
msg

Example 7: Automatic edge device registration

Registers new OPC server information onto edge devices in bulk when it arrives.

[flow_on_webhook] (POST 본문에 OPC + 태그 목록)
   │
   ▼
[flow_edge_opc_create]
   │
   ▼ SUCCESS
[flow_split: data.tags 배열]
   │
   ▼
[flow_edge_tag_create]
   │
   ▼ SUCCESS (모든 태그 등록 완료 후)
[flow_edge_opc_start]
   │
   └── SUCCESS → [flow_log: 엣지 디바이스 가동]

Example call body

{
  "type": "WEBHOOK",
  "data": {
    "opc_id":   "OPC-LINE-A",
    "endpoint": "opc.tcp://line-a.local:4840",
    "tags": [
      { "tag_id": "MOTOR-001.SPEED", "address": "ns=2;s=Motor1.Speed" },
      { "tag_id": "MOTOR-001.TEMP",  "address": "ns=2;s=Motor1.Temp"  }
    ]
  }
}

Example 8: Improving external API reliability (Retry)

Applies back-off retries to an intermittently failing external API call and notifies the operator once the maximum attempts are exceeded.

[flow_on_asset_event]
   │
   ▼
[flow_http_request: 외부 ERP API]
   │
   ├── SUCCESS  → [flow_log]
   └── FAILURE  ─▶ [flow_retry]
                       │
                       ├── SUCCESS    ─▶ (다시 flow_http_request 로 루프 결선)
                       └── EXHAUSTED  ─▶ [flow_send_email: 'ERP 동기화 N회 실패']

Retry node settings

• max_attempts: 5
• backoff_ms: 2000
• backoff_multiplier: 2.0 → intervals of 2s, 4s, 8s, 16s, 32s

Example 9: Plugin event routing (low OEE notification)

Sends a push notification to the line manager when the OEE value falls below a threshold.

[flow_on_oee_event]
   │
   ▼
[flow_script_filter: data.availability * data.performance * data.quality < 0.6]
   │
   ├── TRUE
   ▼
[flow_template: '라인 ${originator.id} OEE ${data.oee_pct}%']
   │
   ▼
[flow_send_push: 라인 매니저]

Template example

• body: 라인 ${originator.id} OEE ${data.oee_pct}% (목표 60% 미달, 주요 손실: ${data.top_loss})

Example 10: Message refinement pipeline (Throttle + Debounce)

Limits high-frequency tag changes to once per minute and additionally fires downstream nodes only when there has been no change for 5 seconds.

[flow_on_tag_point]
   │
   ▼
[flow_throttle: max_msgs=1, window_ms=60000]
   │
   ├── SUCCESS    → [flow_debounce: window_ms=5000]
   │                                │
   │                                ▼
   │                          [flow_save_attributes]
   └── THROTTLED  → [flow_log: 차단됨]

Example 11: Combining multiple triggers (Merge)

Bundles three kinds of events — OEE/RAM/EMS — in 30-second windows and sends a single report message.

[flow_on_oee_event]  ─┐
[flow_on_ram_event]  ─┼─▶ [flow_merge: window_ms=30000]
[flow_on_ems_event]  ─┘                │
                                       ▼
                              [flow_script_transform: 요약 메시지 생성]
                                       │
                                       ▼
                              [flow_send_email: 일일 요약]

If you place several trigger nodes in the same flow, all of them act as entry points. All messages that arrive during the window accumulate in the data.merged array of flow_merge.

Example 12: Bundling common processing into a subflow

Separates common message refinement logic (duplicate check + unit conversion + storage) into a dedicated flow and calls it from several triggers.

Main flows (each independent)

[flow_on_tag_point]   ─▶ [flow_subflow: target_flow_id=FLOW_00099]
[flow_on_asset_data]  ─▶ [flow_subflow: target_flow_id=FLOW_00099]

Subflow FLOW_00099

(트리거 없음 — 호출 전용)
[flow_log: 'subflow in']
   │
   ▼
[flow_check_existence_field: data.value 존재]
   │
   ├── TRUE
   ▼
[flow_script_transform: 단위 변환]
   │
   ▼
[flow_save_tag_point]

A subflow can be saved even without a trigger and can be used exclusively for external invocation (a warning appears on save).

Example 13: Automatic daily report at shift change

At the end of the night shift each day (for example 06:00), emails a summary of production, quality, and downtime spanning yesterday through today.

[flow_schedule: 0 0 6 * * ?]   (매일 06:00)
   │
   ▼ TIMER
[flow_http_request: 내부 통계 API /api/report/daily]
   │
   ▼ SUCCESS
[flow_script_transform: 본문 마크다운 생성]
   │
   ▼
[flow_send_email]

Script Transform example

const r = data.report;
data.subject = `[${r.site_id}] ${r.date} 일일 운영 요약`;
data.body    =
  `■ 생산: ${r.qty_done}/${r.qty_planned} (${(100*r.qty_done/r.qty_planned).toFixed(1)}%)\n` +
  `■ 가동률: ${r.availability}%\n` +
  `■ 품질률: ${r.quality}%\n` +
  `■ 다운타임 Top3:\n` +
  r.downtimes.slice(0,3).map(d => ` - ${d.code} ${d.minutes}분`).join('\n');
msg

Example 14: Automatic downtime aggregation

When an asset STOP event arrives, updates the accumulated downtime per shift and sends a notification when a threshold is exceeded.

[flow_on_asset_event]
   │
   ▼
[flow_msg_type_filter: type in [ASSET_EVENT]]
   │
   ▼ TRUE
[flow_script_filter: data.event_type === 'STOP']
   │
   ▼ TRUE
[flow_script_transform: 다운타임 분 단위 계산]
   │
   ▼
[flow_save_attributes: 자산 누적 다운타임 갱신]
   │
   ▼ SUCCESS
[flow_script_filter: data.shift_downtime_min > 30]
   │
   ▼ TRUE
[flow_send_push: '시프트 다운타임 30분 초과']

Example 15: Automatic isolation of a defective quality line

When 5 or more consecutive defects are reported during quality inspection, issues a stop command to the line asset and suspends the work order.

[flow_on_asset_event]   (event_type=QUALITY_FAIL)
   │
   ▼
[flow_script_transform: data.consecutive_fail = (... + 1)]
   │
   ▼
[flow_save_attributes]
   │
   ▼ SUCCESS
[flow_script_filter: data.consecutive_fail >= 5]
   │
   ▼ TRUE
[flow_publish_asset_command: cmd_key='STOP']
   │
   ▼
[flow_abort_work_order: abort_code='QUALITY']
   │
   └── SUCCESS → [flow_send_email: 품질 매니저 + 라인 매니저]

Example 16: Energy threshold exceeded — recommend pausing the line

When a line's hourly energy consumption exceeds the budget, sends an SMS with a recommendation message to the operator.

[flow_on_ems_event]
   │
   ▼
[flow_script_filter: data.power_kwh > data.budget_kwh * 1.2]
   │
   ▼ TRUE
[flow_template: '${originator.id} 시간당 ${data.power_kwh}kWh (예산 ${data.budget_kwh}kWh 초과)']
   │
   ▼
[flow_send_sms]
   │
   └── SUCCESS → [flow_save_attributes: 자산에 마지막 경고 시각 기록]

Example 17: Bidirectional synchronization with an external system — work order state mirroring

Bidirectionally synchronizes states arriving from an external ERP with internal work order states.

Downstream (ERP → internal)

[flow_on_mqtt_subscribe: erp/work-order/status]
   │
   ▼
[flow_script_transform: 메시지 → 도메인 매핑]
   │
   ▼
[flow_switch: data.status]
   │
   ├── case "STARTED" ─▶ [flow_start_work_order]
   ├── case "PAUSED"  ─▶ [flow_pause_work_order]
   ├── case "DONE"    ─▶ [flow_end_work_order]
   └── DEFAULT        ─▶ [flow_log: WARN]

Upstream (internal → ERP)

[flow_on_entity_event]   (originator.entity_type=Order)
   │
   ▼
[flow_msg_type_filter: type in [ENTITY_UPDATED]]
   │
   ▼ TRUE
[flow_template: ERP 형식으로 변환]
   │
   ▼
[flow_mqtt_publish: erp/work-order/status]

To avoid infinite loops in bidirectional synchronization, mark the message source with a key such as metadata.source and filter out self-published messages at the trigger stage.

---

Import/Export

You can serialize flow definitions to JSON to port them to another environment or back them up.

Action	Location	Description
Export	내보내기 button at the top of the edit screen	Downloads the entire graph + nodes + wires + node settings as JSON
Import	가져오기 button at the top of the list screen	Paste or upload JSON text
Automatic snapshot	Automatic on save	Stored per version when the graph is saved (for rollback)

Import behavior

• A new flow ID is always issued (prevents overwriting an existing ID)
• Node IDs are also reissued and wire links are automatically remapped
• A flow immediately after import is stored in the undeployed state; the operator must review and deploy it manually

---

Error Handling

Per node

• If an exception occurs during node processing, the message is automatically routed to the FAILURE relation.
• If no node is connected to the FAILURE output, the message is dropped and only an error log remains.
• Messages that do not match a trigger node's *_pattern are treated as SKIPPED, are not passed to downstream nodes, and are not included in the processing/error counters.
• All errors are shown in the live debug panel and the execution history screen.

Per flow

Item	Default	Description
max_depth	100	Limit on the cumulative number of node visits while processing one message
max_revisit	3	Limit on revisits to the same node (prevents infinite cycles)
flow_timeout_ms	30,000	Forcibly terminates when the processing time is exceeded

External IO node retries

External IO nodes such as HTTP, Kafka, and external DB can retry immediately inside the node via the retry_count / retry_delay_ms options, and are routed to the FAILURE relation when all retries fail. If you need finer back-off or EXHAUSTED branch handling, use a separate flow_retry node.

---

Operational Diagnostics

Administrators can check the state of the flow engine's dispatch queue (whether workers are running, waiting queue size, cumulative processed/failed/dropped counts, last error) from the System menu.

The server periodically inspects the load of the flow worker pool in the background, and if the load persists beyond a certain time, it records a one-line state change (HEALTHY → DEGRADED → CRITICAL) in the operations log. When it recovers to normal, one more recovery log line is written.

---

Permissions

Flow does not introduce a separate permission model; it uses the existing system authentication/authorization as-is.

Feature	Required permission
Viewing the list / viewing execution history	All authenticated users
Creating, editing, deploying, deleting flows	ADMIN
Import/export / Redeploy All	ADMIN

---

Operational Patterns (Recipes)

A library of frequently used wiring shapes. You can copy each pattern as-is and use it as the starting point for a new flow.

Pattern 1 — Processing + notification branches

Dual branching: store on SUCCESS, notify on FAILURE.

[Action Node]
   ├── SUCCESS → [flow_log] / [flow_save_attributes] / ...
   └── FAILURE → [flow_send_email] / [flow_send_push]

Pattern 2 — Safe retry

Configure back-off + an EXHAUSTED handler so external IO is resilient to transient failures.

[risky_node] ─[FAILURE]→ [flow_retry] ─[SUCCESS]→  (risky_node 로 루프)
                                       └[EXHAUSTED]→ [에러 핸들러]

Pattern 3 — Block load with pre-filtering

Filter messages in advance with *_pattern at the trigger stage to reduce downstream volume. Non-matching messages are treated as SKIPPED and are not counted.

[flow_on_tag_point]    (옵션 tag_id_pattern: "MOTOR-*.SPEED")
   │
   ▼
[필터/변환/액션 ...]

Pattern 4 — Multi-case branching

Branch into several paths depending on state/type.

[flow_switch]   (case별 boolean 표현식)
   ├── case A → [...]
   ├── case B → [...]
   └── DEFAULT → [...]

Pattern 5 — Window accumulation + emit once

Collect high-frequency input over a period and convert it into a single message.

[High-rate Trigger]
   │
   ▼
[flow_merge: window_ms=10000]   → data.merged 배열에 누적
   │
   ▼
[flow_script_transform: 요약]
   │
   ▼
[Action / 외부 발송]

Pattern 6 — Retry + bypass after the retry limit

Once retries are exhausted, divert to a backup path (a different API, a notification, DB storage).

[Primary HTTP] ─[FAILURE]→ [flow_retry]
                            ├─[SUCCESS]    → (Primary HTTP)
                            └─[EXHAUSTED]  → [Backup HTTP] ─[FAILURE]→ [flow_log/Email]

Pattern 7 — Follow-up actions after a domain change

Convert tag-level alarms to asset level and delegate to per-asset processing.

[flow_on_tag_alarm]
   │
   ▼
[flow_change_originator: Tag → Asset]
   │
   ▼
[자산 단위 액션 (Create Work Order / Publish Asset Event 등)]

Pattern 8 — Combining throttle and debounce

Throttle to at most once per minute + process only when there has been no change for 5 seconds.

[High-rate Trigger]
   │
   ▼
[flow_throttle: max_msgs=1, window_ms=60000]
   │
   ▼ SUCCESS
[flow_debounce: window_ms=5000]
   │
   ▼
[Action]

Pattern 9 — Modularizing common logic with a subflow

When several entry points must share the same downstream processing (validation, refinement, storage), separate it into a subflow.

메인1: [Trigger A] → [flow_subflow: target=FLOW_99]
메인2: [Trigger B] → [flow_subflow: target=FLOW_99]

서브 (FLOW_99): (트리거 없음)
[검증] → [정제] → [적재]

Pattern 10 — Bypassing direct triggers

To feed one flow's result into the next flow's input, publish to a domain channel with flow_dds_publish and have the other flow receive the same message type as its trigger.

플로우 A: [...] → [flow_dds_publish: type=ASSET_EVENT, originator=...]
플로우 B: [flow_on_asset_event] → [...]

---

Use Cases (At a Glance)

Scenario	Configuration
MES integration	flow_schedule → flow_http_request → flow_script_transform → flow_create_work_order
Relaying events externally	flow_on_asset_event → flow_msg_type_filter → flow_mqtt_publish
Data refinement & storage	flow_on_tag_point → flow_script_transform → flow_save_tag_point
Alarm automation	flow_on_tag_alarm → flow_script_filter → flow_send_email + flow_create_work_order
External DB synchronization	flow_jdbc_poll → flow_script_transform → flow_create_asset
Webhook reception	flow_on_webhook → flow_script_filter → flow_publish_asset_command
Automatic alarm band adjustment	flow_on_asset_aggregation → flow_script_transform → flow_update_tag_alarm_band_numeric
Work order state automation	flow_on_asset_event → flow_switch → flow_start/end/pause/resume_work_order
API reliability improvement	flow_http_request ─FAILURE→ flow_retry ─SUCCESS→ loop / EXHAUSTED→ notification
Low OEE notification	flow_on_oee_event → flow_script_filter → flow_template → flow_send_push
Bulk edge registration	flow_on_webhook → flow_edge_opc_create → flow_split → flow_edge_tag_create → flow_edge_opc_start
High-frequency refinement	flow_throttle → flow_debounce → downstream
Combining multiple events	Multiple triggers → flow_merge → summary transform → send

---

Step-by-Step Debugging Guide

The standard procedure to follow when a flow does not behave as intended.

Step 1 — Check whether it fired

On the list screen, see whether the execution count for that flow is increasing.

Observation	Meaning / next action
Count is 0	The trigger is not firing. Check the flow's Deploy state, trigger node wiring, and the *_pattern option
Count increases but errors increase too	Failure at an action node. Go to step 2
Count increases but downstream processing does not happen	Missing branch wiring. Check that all outputs such as FAILURE/THROTTLED/EXHAUSTED are handled

Step 2 — Check per-node flow with live debug

Open the edit screen and turn on the live debug panel (2-second refresh).

Observation	Meaning
A node's indicator stays gray	The message is not reaching it — it branched to FAILURE at the preceding node or was blocked by a filter
Red indicator + ERROR line	An exception during node processing. Check the message's data.error field and the NODE_ERROR event in /flow/log
Green indicator but the next node is gray	Output relation label mismatch. Check that the wire label exactly matches the node's output relation (SUCCESS/TRUE, etc.)

Step 3 — Trace the message on the execution history screen

On the /flow/log screen, trace the full flow of a single message by message ID.

• Enter the msg_id you saw in live debug into the Message ID filter
• Sorted chronologically in the order FLOW_START → NODE_IN → NODE_OUT → FLOW_END
• If you see NODE_ERROR, that node's error_message is the cause

Step 4 — Inspect node settings

Common mistakes:

Mistake	Check
Typo in the *_field path	Is it data.tag_id? Or metadata.tag_id? Verify the actual keys in the live debug message preview
${...} template variables not substituted	Whether the variable path exists in the message and has no typos
External IO timeout	Increase timeout_ms. Also check the external system's own response time
Permissions / auth headers	Whether the headers JSON format is correct and the token is still valid

Step 5 — Isolation test

Move the problem node alone into a new temporary flow and use Test Run to fire a single message, verifying the result in isolation. If it works correctly, suspect the original flow's wiring or the shape of the message from the previous stage.

Step 6 — Reproduce after resetting counters

Reset all counts, then fire a single message so you can reproduce the problem with clean statistics — this often makes the problem much clearer.

---

Common Problems

Symptom	Cause / action
The trigger does not fire	Check whether the flow is deployed and whether the trigger node's *_pattern is too narrow. For schedule/external subscription nodes, run Redeploy All and try again
The live debug panel is empty	Check whether the message counter is increasing — trigger pattern non-matches (SKIPPED) are not counted. Also check whether the pause toggle at the top of the panel is on
Counter values accumulate abnormally	Reset the statistics window with the Reset All Counts button at the top right of the node settings
The same message is processed repeatedly	Suspect a cycle. Revisits to the same node are allowed only up to the max_revisit default of 3. Separate with flow_subflow, then adjust the input volume with flow_throttle/flow_debounce
External system calls fail intermittently	Wire back-off retries using the retry_count/retry_delay_ms node options or a flow_retry node
Schedules do not work after import	Run Redeploy All at the top of the list
Some fields are unchanged after running an Update node	This is intended. Update nodes update only the fields you entered and keep the rest. For a full update, use Delete + Create
A Create node fails with a NOT NULL error	Check in the node settings that the *_field dynamic option path actually exists in the message. Some NOT NULL columns are auto-filled, but key IDs (asset_id, tag_id, etc.) must be provided yourself
flow_kafka_publish / flow_mqtt_publish does not publish	Check the external broker connection information (bootstrap_servers/broker_url) and topic permissions. Wire a flow_log alongside to confirm the message reaches the point just before publishing
flow_http_request does not respond	Increase timeout_ms (default 5000) and specify the body explicitly in body_template in ${msg.data.x} form. The response is stored in data.response_status / data.response
flow_retry does not retry	This is a wiring mistake. The SUCCESS output of flow_retry must loop back to the original failing node for retries to occur (see Pattern 2 below)
The same rows keep coming back in flow_jdbc_poll	The polling SQL's WHERE condition must include a status update after processing (for example WHERE sync_status='NEW' plus flow_jdbc_query to UPDATE ... sync_status='OK' at the end of the same flow)
Cannot find a direct alarm trigger node	This is an intentional exclusion. Alarms must be raised only through the flow_create_alarm_config path so that alarm history stays consistent
A script node fails with a Java class access error	It is blocked by the script sandbox. Wire a separate external integration node for external calls
A work order state transition node only returns FAILURE	The current state is not a valid starting state for the transition. For example, flow_pause_work_order works only from the START state. Check the state beforehand with flow_check_existence_field / flow_script_filter
Test Run worked but the graph was left changed	Test Run always fires the saved version. To verify your changes, save first, then run the test
An imported flow is inactive	This is intended. Review it and turn on the Deploy toggle yourself

---

Frequently Asked Questions (FAQ)

Questions operators frequently ask when first using Flow.

Q. Can one flow have multiple triggers? A. Yes. If you place several trigger nodes in the same flow, all of them become entry points and fire independently. Combined with flow_merge, you can merge several kinds of events into a single downstream process.

Q. Can I create a flow with no trigger? A. Yes. A warning appears on save, but the save succeeds. You can use it as a "library" that is fired only by Test Run or by a flow_subflow call from another flow.

Q. Can one message pass through several nodes at once? A. Yes. If you connect several wires to a node's output port, the same message branches simultaneously and is delivered to all downstream nodes.

Q. Can a transform node create the message itself and change it into a different message? A. Yes. In flow_script_transform you can freely change msg.type, msg.originator, msg.data, and msg.metadata. However, even if you make the message look like a different trigger type, it will not automatically invoke another flow (the original trigger mapping is preserved). To call another flow, use flow_subflow or flow_dds_publish.

Q. How many automatic snapshots are kept? Can I restore one myself? A. A snapshot is stored automatically on each graph save, and the retention policy follows the operating environment configuration. There is no UI for restoring one directly; if needed, ask an administrator to roll back to a specific version. The safest approach is to keep a JSON file externally via 내보내기 before making changes.

Q. How long is execution history retained? A. 7 days by default. If you need long-term retention, ship it to an external log system with flow_kafka_publish or flow_jdbc_query.

Q. I want to reset the statistics of only one flow. A. Use Reset error counts (errors only) or Reset all counts (everything) at the top right of the node settings on the edit screen. Other flows are unaffected.

Q. I want to move a flow to another environment (dev/staging/production). A. Download the JSON with 내보내기 on the edit screen and upload it with 가져오기 on the target environment's list screen. An imported flow is always stored with a new ID and in the undeployed state, so the operator can review it and deploy it manually.

Q. The same payload is being processed twice. How do I prevent that? A. Narrow it down with *_pattern at the trigger stage, or limit the frequency with flow_throttle/flow_debounce. For external input (webhook, MQTT), have the sender include an idempotency key and filter duplicates with flow_check_existence_field or a message-ID-based filter.

Q. Is cyclic (loop) wiring safe? A. Intended cycles such as flow_retry are safe. Unintended cycles are automatically blocked by max_revisit (3 times by default). Still, during operation it is a good idea to visualize the flow with flow_log so you catch unintended cycles early.

Q. Are flow changes reflected in real time? A. They take effect immediately on graph save. However, triggers with their own scheduler — schedules, external subscriptions, external DB polling — must be re-registered via Redeploy All or by toggling the flow undeploy → deploy.

Q. Can I change the ID of an already registered node? A. Node IDs are assigned automatically by the system. You can change the display name in the node configuration form, and the display name is used in live debug and execution history.

Q. Could a user without permission change a flow by mistake? A. Creating, editing, deploying, and deleting flows requires ADMIN permission. Ordinary users can only view the list and the execution history.

---

Operational Best Practices

Principles worth following to operate flows safely in production.

Load management

• Pre-filter with *_pattern at the trigger stage to reduce downstream volume. Tag points arriving tens of thousands of times per minute are the most expensive entry point.
• Control load on external systems by combining external calls (flow_http_request/flow_jdbc_query, etc.) with flow_throttle / flow_debounce.
• If many downstream nodes must hang off the same input, separate them with flow_subflow to reduce the total node count in the graph. The more nodes there are, the higher the live debug polling cost.
• Turn on debug mode only during the operational verification stage. The more you store, the faster the 7-day execution history retention window fills up.

Writing safely

• Always wire a FAILURE branch on every action node (at minimum a flow_log). If FAILURE is empty, messages are silently dropped and incidents become hard to trace.
• Combine external IO nodes with flow_retry where possible to absorb transient failures.
• Update nodes update only the fields you entered (partial update). For a full overwrite, use a Delete + Create combination.
• A graph with no trigger node is for manual test runs only. Check the warning on save so you do not omit a trigger by mistake.
• Create cycles (wiring that returns to the same node) only in intended forms such as flow_retry; otherwise max_revisit protects you, but visualize suspicious graphs with flow_log.

Change management

The recommended sequence when changing a flow that is in operation.

1. Back up — download the current graph as a JSON file with 내보내기 on the edit screen and keep it (automatic snapshots are also stored, but external storage is safer).
2. Work in a copy or a new flow — rather than modifying the running flow immediately, make and verify changes in a new flow (or a copy created via import).
3. Test run — inject JSON messages directly to fire every branch (SUCCESS/FAILURE/EXHAUSTED, etc.) at least once and check the results in the execution history.
4. Roll out to production — 내보내기 the verified flow's graph JSON → 가져오기 in the production environment → operator reviews and turns on Deploy.
5. Roll back if there is a problem — disable immediately with the Undeploy toggle. Since automatic snapshots are stored, a developer can revert to a previous version on request.

Making use of counters and statistics

• If the processing-trend sparkline on the list screen suddenly flattens, check the possibility of the trigger not firing or SKIPPED handling.
• If the error-ratio donut chart looks abnormal, suspect the *_field path in the node settings. Failures are common when the key is absent from the message payload.
• After operational verification, reset the statistics window with Reset All Counts to re-measure your normal baseline.

---

Emergency Response Procedures

Step-by-step procedures to use quickly when a problem occurs during operation.

Scenario 1 — A specific flow runs away (message explosion)

Symptom: a flow's execution count spikes tens to hundreds of times above normal, with errors increasing too

Actions

1. Immediately disable that flow with the Undeploy toggle (1 second on the list screen)
2. Identify which trigger caused the runaway from the live debug panel and execution history
3. Narrow the trigger node's *_pattern option, or wire a flow_throttle / flow_debounce immediately after it
4. If necessary, restrict the message type with flow_check_existence_field or flow_msg_type_filter
5. After fixing, reproduce with Test Run → confirm normal behavior → resume Deploy

Scenario 2 — Bulk failures due to an external system outage

Symptom: errors accumulate simultaneously on external IO nodes such as HTTP/Kafka/external DB

Actions

1. If the impact is broad, undeploy the affected flows in bulk (select them all on the list screen and undeploy at once)
2. Confirm the external system has recovered
3. If an external IO node has no flow_retry wiring, add it
4. If the external system has become slow, adjust timeout_ms
5. Depending on the environment, run Redeploy All and then resume Deploy

Scenario 3 — Infinite loop caused by a cycle

Symptom: a single message keeps passing through the same node, with processing time accumulating

Actions

1. max_revisit protection blocks it automatically after at most 3 revisits, but undeploy and inspect for operational safety
2. Trace the cycle visually in the graph (follow the wires on the edit screen)
3. If it is an intended loop (flow_retry), check that max_attempts is appropriate
4. If the cycle is unintended, remove the wiring or add a branch with flow_check_relation
5. After fixing, Reset All Counts → redeploy

Scenario 4 — Suspected data corruption (incorrect automatic updates)

Symptom: automation updated domain data in an unintended way

Actions

1. Immediately undeploy that flow
2. Check the previous version from the externally stored graph JSON backup or from an automatic snapshot (with administrator assistance)
3. Analyze the graph: check for unintended Update node wiring, wrong *_field paths, and script transform errors
4. Correct the affected domain data through a separate procedure on the back-office screens
5. Verify the corrected graph with Test Run, then redeploy

Scenario 5 — Stopping during system inspection/maintenance

Symptom: you want to pause external IO calls during an external system maintenance window

Actions

1. Undeploy the affected flows in bulk (multi-select on the list)
2. Resume Deploy after the maintenance ends
3. For triggers with their own scheduler (schedule, external MQTT subscription, external DB polling), run Redeploy All once and confirm they are registered correctly

Scenario 6 — Live debug storage queue is full

Symptom: the storage queue size on the operational diagnostics page is near the threshold, with the drop count increasing

Actions

1. Reduce the number of flows with debug mode on and the volume stored (turn debug mode OFF for flows you have finished verifying)
2. Reduce downstream volume itself with *_pattern at the trigger stage
3. When the system enters automatic recovery mode, a DEGRADED/CRITICAL log line is written by the load watchdog — share it with the administrator

In every scenario the priority order is undeploy immediately → identify the cause → fix → verify → redeploy. Also refer to the change procedure (Change management).

---

Security & Sensitive Data Handling

Flows handle sensitive input and output — external API calls, email/SMS delivery, webhook reception — so please observe the following principles.

Tokens and credentials

• Do not enter API tokens or passwords directly into node settings in plaintext. Use environment variable substitution in the form ${ENV_VAR} so they are injected from the production environment's secret store.
• To avoid exposing tokens in production, isolate the token position in the headers JSON into environment variables as much as possible.

// 권장
{ "Authorization": "Bearer ${MES_TOKEN}" }

// 비권장
{ "Authorization": "Bearer eyJhbGciOi..." }

Protecting webhook entry points

flow_on_webhook is an internal entry point called by authenticated users, but when you expose it externally, always wire a payload validation filter.

// flow_script_filter — 토큰 + 필수 필드 검사
if (data.token !== '${WEBHOOK_TOKEN}') return false;
if (!data.asset_id || !data.cmd_key)   return false;
true

Masking sensitive data

Message previews are displayed in the live debug panel and execution history. Mask personal information (names, contact details, accounts) and tokens just before storage.

// 디버그/로그 적재 직전 변환
data.email_masked = data.email
  ? data.email.replace(/(.{2}).+(@.+)/, '$1***$2') : null;
delete data.email;
delete data.token;
msg

Handling external IO response bodies

data.response of flow_http_request stores the entire response body. If the response contains sensitive data, extract only the keys you need in an immediately following transform node and remove the original.

// 응답에서 필요한 필드만 보존
data = { id: data.response_obj.id, status: data.response_obj.status };
msg

Script node isolation

Script nodes run in a secure sandbox with file, network, and arbitrary class access blocked. If you need external calls, always wire a separate external integration node.

---

Flow Metrics and Alarms

The indicators collected automatically while a flow is running, where to view them, and how to set alarms on them.

Flow-level KPIs

Each row on the list screen (/flow/index) shows the following KPIs.

Indicator	Meaning	Abnormal signal
Total executions	Number of times a trigger fired and passed through the graph once (successes + failures)	Sharp drop vs. normal → the trigger is dead / sharp rise → runaway
Error count	Number of times a message took the FAILURE branch at any node in the graph	Inspect if it is 5% or more of the total
Error rate	에러 / 전체 × 100	Set an alarm when it exceeds a defined threshold
Last execution	Time of the most recent firing	Only OK if "no firing for 5+ minutes" is normal for that flow
Average processing time	Average ms for one message to pass through the whole graph	Changes significantly when external IO nodes are added/removed
Recent processing time trend	5-minute sparkline — in the live debug panel	On spikes, check external system responses

Node-level KPIs

When you click a node on the edit screen, the following appears at the bottom right of the node.

[Node Name]
처리 999 · 에러 3 · 평균 12ms · 최근 18ms

Location	Display
Top indicator	Gray (idle) / green (processing) / red (error)
Bottom label	처리 N · 에러 M · 평균 Xms · 최근 Yms

The four node counters

Counter	Meaning
Processed	Number of messages that entered the node and left via a normal branch
Errors	Number of FAILURE branches or exceptions
Average processing time	The node's own processing ms (including external IO)
Recent processing time	Processing ms for the most recent message

System-level metrics

Metrics for the flow engine as a whole can be checked on the System → Monitoring screen.

Metrics are exposed as the following five entries under the JMX domain plantpulse.core.engine.

Metric	Kind	Meaning	How to read it
FLOW_EXECUTOR_QUEUE_SIZE	Gauge	Waiting queue size of the flow execution pool	If it keeps growing, processing cannot keep up with intake
FLOW_EXECUTOR_ACTIVE_COUNT	Gauge	Number of active threads in the execution pool	If it stays at the pool size, the pool is saturated
FLOW_EXECUTOR_DROPPED_COUNT	Gauge (cumulative)	Cumulative number of tasks discarded due to pool saturation	If it is not 0, messages have been lost — the first value to check
FLOW_DEBUG_QUEUE_SIZE	Gauge	Size of the queue waiting to store debug events	Grows when many flows have debug mode on
FLOW_EXECUTION_TIME	Timer	Distribution of processing time per flow	Track average and maximum

There is no metric named flow.engine.*

Older documents listed flow.engine.queue_depth, in_flight, dispatch_lag_ms, exec_p95_ms, failed_per_min, and script_timeout_per_min in a table, but no metrics with those names exist. The five above are all of them.

(flow.engine.enabled is not a metric but a configuration key in engine.properties — the master gate that turns the whole flow engine off.)

Patterns for registering metric-based alarms

Four patterns for monitoring flow metrics with EQL alarms or CEP → Trigger.

Pattern A — Error rate exceeds a threshold

context EVERY_1_MINUTES
SELECT flow_id, count(CASE WHEN status='FAILURE' THEN 1 END) * 100.0 / count(*) AS err_pct
FROM   AssetEvent.win:time(5 min)
WHERE  event_type = 'FLOW_EXEC'
GROUP  BY flow_id
HAVING err_pct > 5

Pattern B — Processing queue backpressure

Alarm when flow.engine.queue_depth > 500 persists for more than a minute — a situation where the trigger firing rate exceeds the processing rate.

Pattern C — A specific flow stops firing

SELECT * FROM pattern [
  every a = AssetEvent(event_type='FLOW_EXEC', flow_id='my-flow')
        -> ( timer:interval(15 min)
             and not AssetEvent(event_type='FLOW_EXEC', flow_id=a.flow_id) )
]

Alarm when a flow that normally fires N times per minute has been silent for more than 15 minutes.

Pattern D — Surge in external IO timeouts

context EVERY_5_MINUTES
SELECT flow_id, node_id, count(*) AS timeout_count
FROM   Log.win:time(5 min)
WHERE  module = 'flow-engine' AND code = 'NODE_TIMEOUT'
GROUP  BY flow_id, node_id
HAVING count(*) > 10

Recommended alarm output channels

Alarm kind	Recommended channel
Error rate / firing stopped (direct operations)	Email + push notification
Queue backpressure / timeout surge (system)	Slack/Teams webhook
Automatic diagnostics (reference)	Diagnostic logs only

⚠️ Never make the alarm flow itself depend on the metrics it alarms on — if the alarm flow dies, no alarm arrives at all. Monitor alarm flows with an external health check from System → Monitoring.

---

Message Processing Semantics and Backpressure

The guarantee level and backpressure behavior when the flow engine processes messages.

Delivery guarantee — at-least-once

The flow engine guarantees at-least-once delivery.

Case	Behavior
Normal processing	Fires once → passes through the graph once → completes once
Engine restart during processing	Messages remaining in the trigger queue are processed again after the next boot
Exception at a node	Transitions only to the FAILURE branch. The message is not lost
External IO timeout	FAILURE branch + automatic retry possible with flow_retry

Possibility of duplicates — this is not exactly-once. If flow_create_* is interrupted midway and retried, the same ID may arrive twice, so use on_duplicate=skip or the external system's idempotency key.

Idempotency key pattern

A pattern for guaranteeing idempotency when a flow calls an external system.

// flow_script_transform — 멱등성 키 생성
msg.data.idempotency_key =
  msg.metadata.asset_id + '|' +
  msg.metadata.ts + '|' +
  msg.data.event_type;

Then attach it as a header on the external call:

"headers": { "Idempotency-Key": "${data.idempotency_key}" }

Processing order — FIFO per trigger

Messages with the same originator	Different originators
FIFO per trigger (in arrival order)	Processed in parallel (order irrelevant)

Even if two events for asset A occur nearly simultaneously, A's events are processed in the order they occurred. Events for asset A and asset B may be processed in parallel on different workers.

Backpressure

Behavior when the processing rate cannot keep up with the firing rate.

Situation	Behavior
Queue depth < 80%	Normal — new triggers are enqueued immediately
Queue depth 80%–100%	Diagnostic WARN log is emitted — the queue still accepts messages
Queue depth = 100% (full)	New triggers are dropped + a diagnostic ERROR log is emitted

What the operator should do when the queue is full

1. Check flow.engine.queue_depth in System → Monitoring
2. Identify the flow whose execution count has spiked on the List screen
3. Narrow that flow's trigger pre-filter (*_pattern) to reduce load
4. Or temporarily undeploy that flow as an emergency measure

Circuit breaker (external IO)

External IO nodes (HTTP/Kafka/MQTT/email) are blocked entirely for 30 seconds when the following conditions are met.

Condition	Threshold
Consecutive failures within the last minute	≥ 10
Average response time	≥ 10 seconds

While blocked, every message entering that node is immediately branched to FAILURE. When the external system recovers, the block is released automatically. This behavior isolates external outages so they do not paralyze the flow engine itself.

The moment the circuit opens is recorded in the diagnostic log as code = CIRCUIT_OPENED. If the external system recovered quickly, you can look at metadata.retry_count in the Execution history and manually re-run the missed messages.

Cycle prevention

A cycle like the following inside a flow can cause an infinite loop.

A → B → C → A  (잘못된 결선)

The flow engine blocks cycles with two lines of defense.

Defense	Behavior
max_depth=100	Forcibly terminates after passing through 100 nodes
max_revisit=3	Discards the message and logs a diagnostic ERROR the moment it visits the same node a fourth time

The SUCCESS branch → original node loop of flow_retry is an intentional cycle, and since metadata.retry_count increases along with it, it terminates normally before max_revisit triggers.

---

What You Will See in Execution History

The "error code catalog" in older documents never existed

Codes such as NODE_SCRIPT_TIMEOUT, FLOW_TIMEOUT, FLOW_MAX_DEPTH, TRIGGER_QUEUE_FULL, WEBHOOK_AUTH_FAIL, and DOMAIN_DUP_KEY were listed in a table, but nothing anywhere in the product produces those strings. Searching the logs for those codes will return zero results forever.

What the flow engine actually records are the five event types below.

Five kinds of events are stored in the flow execution history (mm_flow_log, left menu Automation > Flow Execution History).

Event type	When	Storage condition
FLOW_START	On entering the flow	Always
FLOW_END	On flow processing completion (both success and timeout)	Always
NODE_IN	When a node receives a message	Only in debug mode
NODE_OUT	When a node emits a message	Only in debug mode
NODE_ERROR	Exception during node processing	Always

The fields every row carries:

Field	Contents
level	INFO / WARN / ERROR
flow_id · flow_node_id · node_name · node_type	Which node of which flow
msg_id	For per-message tracing — group by this value to follow one message's path
message · error_message	Human-readable description, exception message
duration_ns	Node execution time (nanoseconds) — meaningful only in NODE_OUT and NODE_ERROR

Node-level logs are not recorded under normal conditions

NODE_IN and NODE_OUT are stored only in debug mode. Seeing only FLOW_START and FLOW_END in the execution history under normal conditions is expected. To follow node by node, turn on debug mode for that flow — but the log volume will increase considerably.

Finding problems by symptom

Start from symptoms instead of codes.

Symptom	Where to look first
The trigger fires but no node seems to run	Whether the flow is deployed (toggle ON) · whether the trigger's *_pattern is filtering the message out
Processing stops midway	One message exceeded 30 seconds (flow_timeout_ms). Confirm with the FlowExecutor timeout warning in the server log
The same message is processed repeatedly	A cycle. It is blocked automatically by max_revisit (3 times), but inspect the wiring
The graph is so long it never finishes	max_depth (100) exceeded. Split it into subflows
A node throws an exception	Look at error_message on the NODE_ERROR row in the execution history
An external call fails	Check that node's data.response_status and data.error in a downstream node

The limit values (100 · 3 · 30,000 ms) are the defaults of ExecutionState.

---

Cluster & High Availability (HA) Behavior

When the Platform is installed as a cluster, the flow engine operates in a distributed fashion according to the following rules.

Node role separation

Role	Behavior
Leader	The single node that serially processes flow graph changes (save/deploy/undeploy)
Worker	A general node that fires triggers and processes messages in parallel (every node also acts as a worker)
Scheduler	Responsible for evaluating flow_schedule cron expressions — the same node as the leader

The leader node is elected automatically at platform boot, and if the leader goes down, one of the other nodes is automatically promoted to leader. Operators do not need to designate it manually.

Message distribution — consistent routing per asset

Distribution key	Behavior
originator.id (asset/tag/order ID)	Messages for the same asset are always routed to the same worker — sequence guaranteed
External intake (flow_on_webhook, etc.)	Round-robin distribution

This rule prevents events for asset A from being processed simultaneously on multiple workers and getting out of order. The result is single-worker processing guaranteed per asset while parallel processing holds across assets.

Behavior on leader failure — failover

[Leader 다운 t=0]
        ↓
[다른 노드가 리더 승격 시도 t=0~3s]
        ↓
[새 리더 확정 t=3~5s]  ← cron 스케줄·플로우 배포 변경 재개
        ↓
[기존 워커들은 정상 동작 유지 — 트리거 처리 영향 없음]

Phase	Impact
0–3 s	flow_schedule firing is paused / trigger processing is unaffected
3–5 s	New leader confirmed, scheduling resumes
After 5 s	Normal

Cron firing immediately re-executes one missed firing according to the misfire policy. During operation, if flow_schedule's evaluation granularity is under 5 seconds, you may see brief gaps.

Trigger queue durability

Item	Behavior
Trigger queue location	In-memory queue + persistent storage (transaction log)
On node restart	Unprocessed messages remaining in the queue are dispatched again after the next boot
Node goes down mid-processing	That message is reprocessed, so external systems need idempotency keys due to at-least-once guarantees

Operator checklist for cluster deployment

1. Time synchronization (NTP) on all nodes — so trigger firing times agree across nodes
2. Place external systems (MQTT/Kafka brokers) at network locations reachable by all nodes
3. SMTP settings for flow_send_email are registered once in the system settings — shared by all nodes
4. Tune the worker pool size (flow.engine.workers) to each node's CPU core count

Single-node mode (development & small scale)

• Leader, worker, and scheduler all run in one process
• No failover — if the node goes down, flows stop entirely
• The trigger queue is guaranteed in memory + on disk, so it recovers after restart

---

End-to-End Tracing

How to retrospectively trace how a message passed through the whole graph.

Automatic correlation ID assignment

The flow engine automatically assigns a correlation ID to every trigger-fired message.

{
  "type": "POST_TELEMETRY",
  ...,
  "metadata": {
    "trace_id":   "tr-a1b2c3d4-e5f6-7890-...",
    "span_id":    "sp-01",
    "parent_id":  null,
    ...
  }
}

Field	Meaning
metadata.trace_id	An ID unique to one entire trigger firing — shared by messages passing through every node in the graph
metadata.span_id	A per-node unique ID — updated each time the message passes a node
metadata.parent_id	The span_id of the immediately preceding node

Searching by trace_id on the execution history screen

On the Execution history screen, paste the ID into the trace_id input field to display chronological logs for every node that message passed through.

🔍 trace_id = tr-a1b2c3d4-...

[12:34:56.123] [trg]      flow_on_tag_point      태그=MOTOR.TEMP, value=87
[12:34:56.125] [filter]   flow_script_filter     score>80 → TRUE
[12:34:56.126] [transform] flow_change_originator Tag → Asset
[12:34:56.130] [action]   flow_create_work_order WO-20260513-001 생성
[12:34:56.241] [external] flow_send_email        admin@... 발송 성공

Trace propagation on subflow calls

When you call another flow with the flow_subflow node, the same trace_id is carried through. In other words, the logs of all nodes in the main flow plus the called subflow can be searched under the same trace.

Propagation to external systems

The X-Trace-Id header is automatically attached on external HTTP calls.

GET /api/orders HTTP/1.1
Host: erp.example.com
X-Trace-Id: tr-a1b2c3d4-e5f6-...
X-Span-Id:  sp-04

If the external system receives this header and records it in its logs, you can match logs from both systems by the same ID.

Exposing trace_id in alarms and emails

If you include ${metadata.trace_id} in the alarm body or email template, the operator can trace that incident immediately on the history screen right after receiving the alarm.

제목: [긴급] 모터 과열 — ${metadata.asset_id}
본문:
시각: ${metadata.ts}
값: ${data.value}°C
trace: ${metadata.trace_id}
이력 보기: https://platform.example.com/flow/log?trace_id=${metadata.trace_id}

Trace retention period

Data	Retention period
trace_id and per-node span logs	7 days (same as execution history)
External storage (long-term retention)	Use external system mirroring from Auditing & History Tracking

If the trace_id makes messages too large, you can use flow_script_transform to create a short form exposing only the last 4 characters (like tr-...d4) for alarms and emails. Note that the full ID is required for searching.

---

Graph Pattern Catalog

Ten wiring patterns frequently used in flow graphs.

Pattern 1 — Pipeline (simple serial)

[trigger] → [filter] → [transform] → [action]

| Characteristic | The message passes through the nodes in order | | Use case | Threshold exceeded alarm → send email |

Pattern 2 — Fan-out (1 to N)

                     ┌─→ [action 1]
[trigger] → [t] ─────┼─→ [action 2]
                     └─→ [action 3]

| Characteristic | Several actions process one message simultaneously | | Use case | Alarm occurs → email + SMS + Slack + work order creation |

Copies of the same message are distributed to multiple nodes. Each branch is processed independently, and a failure in one branch does not affect the others.

Pattern 3 — Fan-in (N to 1) — Merge

[trigger A] ──┐
[trigger B] ──┼─→ [flow_merge] → [aggregator] → [action]
[trigger C] ──┘

| Characteristic | Collects messages from several triggers within a time window and processes them at once | | Use case | Combine all alarms in 5 minutes into a single daily report |

Pattern 4 — Scatter-Gather (distribute → collect)

[trigger] → [split] ─┬─→ [process] ──┐
                     ├─→ [process] ──┼─→ [merge] → [action]
                     └─→ [process] ──┘

| Characteristic | Splits an array element by element, processes them, then recombines the results | | Use case | Validate 100 external orders in parallel, then report the results in bulk |

Pattern 5 — Switch (conditional branching)

                            ┌─[CRITICAL]→ [긴급 알람]
[trigger] → [flow_switch] ──┼─[WARN]    → [경고 알람]
                            └─[NORMAL]  → [통과]

| Characteristic | Sends one message down different paths by condition | | Use case | Separate processing channels by alarm priority |

Pattern 6 — Retry with Fallback

[risky] ─[FAILURE]─→ [flow_retry] ─[SUCCESS]─→ (다시 risky)
                                  └[EXHAUSTED]→ [fallback action]

| Characteristic | Automatically retries transient failures; handles permanent failures via an alternative path | | Use case | Retry 3 times on external API failure; if it still fails, notify a human |

Pattern 7 — Circuit Breaker (using circuit blocking)

[trigger] → [throttle] → [external_io] ─[SUCCESS]─→ [save]
                                       └[FAILURE]─→ [log only]

| Characteristic | Rate-limits calls with flow_throttle + bulk blocking by the circuit breaker | | Use case | Protecting an overloaded external system |

Pattern 8 — Dead Letter Queue (DLQ)

[main flow] ─[FAILURE]─→ [flow_save_attributes] → 별도 자산에 적재
                                                  ↑
                              운영자가 주기적 점검 후 수동 재처리

| Characteristic | Sends permanently failed messages to a separate store | | Use case | Collection of failed external ERP synchronization messages (for manual reprocessing) |

Pattern 9 — Sliding Window Aggregation

[trigger] → [flow_throttle 60s] → [transform: 누적] → [action]
                                  (마지막 60건만 유지)

| Characteristic | Makes decisions based on the state within a window of the most recent N items | | Use case | Site emergency mode when more than 30 alarms occur in the last 5 minutes |

Pattern 10 — Saga (multi-stage transaction)

[start] → [step1] ─OK→ [step2] ─OK→ [step3] ─OK→ [complete]
              │            │            │
              └─FAIL→[rollback1]        │
                           │            │
                           ┌─FAIL→[rollback1+2]
                                        │
                                        └─FAIL→[rollback1+2+3]

| Characteristic | Handles partial failures of work spanning several external systems with compensation | | Use case | Create work order → reserve asset → sync with ERP → notify worker (cancel all previous steps if one fails) |

Pattern selection guide

Requirement	Recommended pattern
Simple threshold alarm	Pipeline (1)
One event → several channels	Fan-out (2)
Several events → one summary	Fan-in / Merge (3)
Bulk array processing	Scatter-Gather (4)
Branch processing	Switch (5)
External API reliability	Retry (6)
Protecting an external system	Circuit (7)
Retaining failed messages	DLQ (8)
Decisions on the last N items	Sliding (9)
Consistency across several external systems	Saga (10)

---

Flow Testing Best Practices

A testing strategy for changing and deploying flows safely.

Three-stage testing — unit → integration → simulation

Stage	Tool	What is verified
① Unit	Edit screen ▶ Test Run	A single node's behavior (script expressions, external call responses)
② Integration	Same screen, temporary payload + live debug ON	The whole graph sequence and branching
③ Simulation	Deploy + wait for a real trigger (staging environment)	Real message flow and coupling with external systems

A payload library for test runs

For payload examples per trigger that you can paste into the Test Run dialog, see the Payload Examples by Trigger section. Also prepare boundary cases that occur frequently in production.

Boundary case examples

Case	Payload
null field	{"value": null} — is the script null-safe?
Empty string	{"value": ""} — verify whether an empty string is interpreted as 0
Negative number	{"value": -1} — does the threshold check handle the ± sign as intended
Huge number	{"value": 1e20} — overflow/precision
Unicode	{"name": "한글-Émoji-🚀"} — external system encoding

Safe change procedure

1. 기존 플로우를 [내보내기] (JSON 파일 저장)
2. 새 플로우를 사본으로 만듦 (이름 끝에 `_v2`)
3. 사본의 트리거 패턴을 좁혀 일부 자산만 매칭 (예: TEST-* 사이트만)
4. 신구 동시 배포 — 새 버전 데이터 확인
5. 1주일 안정성 검증 후 신 버전을 전체 트리거 패턴으로 변경
6. 구 버전 해제 + 보관

Keeping regression test scenarios

We recommend keeping frequently used scenarios in a text file and always re-running them after a change.

# regression-tests/alarm-to-workorder.json
{
  "case": "고온 알람 → 워크오더 자동 생성",
  "input": {
    "type": "TAG_ALARM",
    "data": { "alarm_band": "HI_HI", "value": 95.0 }, ...
  },
  "expected": {
    "domain_changes": ["WorkOrder.CREATED"],
    "notifications": ["email:admin@example.com"]
  }
}

External system mock patterns

To mock external system responses in a staging environment:

Method	Description
Point flow_http_request's url at a mock server	Separate the production URL into a staging URL variable
Change flow_jdbc_poll's datasource	Change only production DB → staging DB
Set flow_send_email's to_field to fake@example.com	Prevents accidental sends

Load testing after resetting counters

1. Reset counts (all) for the new flow version
2. Fire normal triggers for 5 minutes
3. Check the processing/error counts and average processing time on the List screen
4. If p95 processing time is more than 20% higher than before, analyze the cause and consider a rollback

---

Performance Limits and Tuning

Processing limits of the flow engine and tuning tips.

Default limits

Item	Default	Notes
Node visits per message (max_depth)	100	Forcibly aborted if not finished after 100 nodes
Revisits to the same node (max_revisit)	3	Prevents infinite cycles
Flow processing time (flow_timeout_ms)	30,000 ms	Forcibly terminated if one message takes more than 30 seconds
External IO node timeout	5,000 ms (timeout_ms)	HTTP/Kafka/MQTT, etc.
Script execution timeout	500 ms (timeout_ms)	Per node
Execution history retention	7 days	Automatically expires afterwards

Recommended window sizes for common nodes

Node	Recommended window	Notes
flow_throttle	1,000–60,000 ms	Match the external system's API limits
flow_debounce	500–5,000 ms	When you want only stable values to pass
flow_merge	5,000–60,000 ms	Too short causes fragmentation, too long increases latency
flow_retry back-off	Start 1,000 ms × 2	With 5 retries: 1, 2, 4, 8, 16 seconds

Ways to increase throughput

• Trigger pre-filtering — admit only the messages you need with *_pattern (most effective)
• Simplify the graph with subflows — keep the main graph to branching and routing, and put heavy processing in subflows
• Wire external IO asynchronously — flatten external API load with flow_delay / flow_throttle
• Debug mode only during verification — turn debug mode OFF once operation is stable
• Wire only the categories you need — do not wire heavy nodes such as Edge or external integration into every branch unnecessarily

Message size

A message's data / metadata are JSON-serialized and stored in the execution history. For huge payloads (response bodies of several MB or more), extract only the keys you need with a transform node whenever possible. Message size is proportional to processing cost in history retention, debug display, and subflow calls alike.

---

Auditing & History Tracking

Where to look when tracking flow changes and executions.

Graph change history

• Automatic snapshots — stored per version on each graph save. To revert to a previous state, ask an administrator.
• Modifier / modification time — the Last modified column on the list screen shows the most recent save time. The modifier is recorded based on the system-authenticated user.
• Recording the reason for change is recommended — recording the reason, the person responsible, and a related ticket number in the flow metadata's Description field greatly improves traceability.

Tracking execution events

• Message ID tracing — use the Message ID filter on the execution history screen to view the full flow of a single message (FLOW_START → all NODE_IN/NODE_OUT → FLOW_END) in chronological order.
• CSV export — use the CSV Download button on the execution history screen to export the current results and hand them to an external audit system.

Domain change history

Domain changes performed by a flow's action nodes (flow_create_* / flow_update_* / flow_delete_*) are all delegated to the same domain services, so they are recorded together in the per-domain change history on the back-office screens. The actor ID is recorded as insert_user_id="flow" or similar, distinguishing it from changes made by ordinary users.

Long-term retention via external storage

If you need retention beyond the execution history period (7 days), create a separate flow to ship key events to an external system (Kafka, an external DB, etc.).

[flow_on_tag_alarm]
   │
   ▼
[flow_kafka_publish: topic=audit.alarm.events]

---

New Flow Deployment Checklist

Items to check before deploying a new flow to production.

Graph structure

• Exactly one trigger node (or an intended number) is wired
• The FAILURE output of every action node is handled (at minimum flow_log)
• If there is a cycle, is it intentional wiring such as flow_retry, and is it within max_revisit protection
• Are retry_count/retry_delay_ms or flow_retry wired to external IO nodes

Node settings

• Is the trigger's *_pattern neither too narrow nor too broad
• Do the *_field dynamic option paths actually exist in the message
• Are all NOT NULL fields of Create nodes filled (aside from automatic enrichment)
• Is the Update node's partial update really what you intended

Verification & testing

• SUCCESS branch passed once with a normal payload
• FAILURE branch passed with an abnormal payload (missing required fields, etc.)
• Retry → EXHAUSTED branch passed with an external IO failure case (if applicable)
• INFO/ERROR display in the live debug panel matches expectations
• Are all stages recorded on the /flow/log screen

Operational safety

• Is a backup (graph JSON export) stored externally
• Are the reason for change and the person responsible recorded in the flow description
• If there is notification wiring (Email/SMS/Push), have the recipients been verified
• Is there a plan to monitor the execution history for 5–10 minutes right after deployment

---

Deployment Strategies — Canary / Blue-Green / A·B Testing

Three strategies for releasing a new flow safely. All of them can be implemented with the Platform's built-in features alone (trigger patterns, message dispatch, live debug).

Strategy 1 — Canary (gradual rollout to some assets only)

Apply the new flow to some assets first, observe for a period, then expand to everything.

[1단계 출시]
  새 플로우 v2 — 트리거 패턴: asset_id LIKE 'LINE-1.%'  (1개 라인만)
  기존 플로우 v1 — 트리거 패턴: asset_id LIKE 'LINE-2.%' OR 'LINE-3.%' OR ...

[2단계 확대 (1주일 후 안정 확인)]
  v2 패턴: 'LINE-1.%' OR 'LINE-2.%'
  v1 패턴: 'LINE-3.%' OR 'LINE-4.%'

[3단계 전체 (2주 후)]
  v2 패턴: '%'   ← 모든 자산
  v1 해제·보관

Canary progress checklist

Period	Monitoring items
Day 1	Error rate < 1% / average processing time within ±20% of before
Week 1	External system integration 100% normal / alarm frequency appropriate
Week 2	Cumulative statistics / user feedback / decision to expand to the next line

Choose a canary line that is among the least impactful of the running lines (for example, one with frequent maintenance or one that runs only at night).

Strategy 2 — Blue-Green (run old and new together, then switch instantly)

[Blue (현재)]                       [Green (새 버전)]
플로우 v1 — 배포됨                  플로우 v2 — 배포 + 격리된 originator
모든 트리거 처리                    metadata.test_mode=true 인 메시지만 처리

[전환 결정 시점]
   v2 트리거 패턴을 v1 과 동일하게 변경 (1초)
   v1 해제 토글 (1초)

The essence of Blue-Green is switching instantly while both versions are deployed — if you find a problem, redeploy v1 immediately to roll back.

v2 isolation wiring pattern

[trigger 모든 메시지]
   ↓
[flow_script_filter — metadata.test_mode === true]
   ↓ TRUE
[새 로직]

Send test messages via the /flow/{id}/run API, specifying metadata.test_mode=true explicitly.

Strategy 3 — A/B testing (comparing performance and outcomes)

Both versions receive the same messages, take different actions, and their results are compared.

[trigger 메시지]
   ↓
[flow_split (메시지 복제)]
   ├ A 경로 → 기존 v1 액션 → [flow_save_attributes target=stats_v1]
   └ B 경로 → 새 v2 액션  → [flow_save_attributes target=stats_v2]

Then compare the cumulative results of stats_v1 vs stats_v2 on the Daily statistics screen.

Automated A/B comparison analysis

-- EQL 으로 두 버전 비교 (예: 알람 생성 누적)
context EVERY_1_HOURS
SELECT
  count(CASE WHEN metadata.flow_version='v1' THEN 1 END) AS v1_count,
  count(CASE WHEN metadata.flow_version='v2' THEN 1 END) AS v2_count
FROM AssetAlarm.win:time(1 hour)

Strategy selection guide

Situation	Recommended strategy
First release of a new automation scenario	Canary — verify on one line, then expand
Major change to existing logic (structural rework)	Blue-Green — instant rollback possible
Measuring which of two algorithms is better	A/B testing
Simple option value adjustment	Change directly — note it in metadata.audit_diff and monitor for a week

Rollback procedure (common)

Roll back immediately when you find a problem:

1. Turn the Undeploy toggle on the new version from the List screen
2. (For Canary/A·B) change the trigger pattern so it matches zero messages
3. Monitor for 5 minutes that the existing version operates on its own
4. Check that there is no code=FLOW_NOT_DEPLOYED in the diagnostic logs
5. Analyze the cause — trace failed messages by trace_id in the Execution history

Pre-release checklist (condensed)

Just the essentials of the New Flow Deployment Checklist on one screen:

[ ] 테스트 실행으로 정상·경계·실패 시나리오 모두 통과
[ ] 외부 IO 노드에 timeout_ms / 재시도 정책 설정됨
[ ] 자격 증명은 ${creds.*} 참조 (평문 미포함)
[ ] 트리거 패턴이 의도한 자산만 매칭
[ ] 영향받는 도메인 (자산/태그/주문) 식별됨
[ ] 운영 시간(특히 야간) 영향 검토됨
[ ] 롤백 시점·기준·담당자 결정됨
[ ] [감사 로그](#audit) 에 변경 의도 메모 작성됨

---

Node Quick Configuration Reference

A cheat sheet collecting only the key settings of nodes used frequently during operation.

Trigger quick settings

Node	Key options
flow_schedule	cron (for example 0 */5 * * * ? = every 5 minutes), or interval_ms
flow_on_webhook	No options — fired externally via POST /flow/webhook/{flow_id}
flow_on_mqtt_subscribe	broker_url, topic, client_id, username/password
flow_jdbc_poll	dsn, sql, interval_ms
flow_on_* (domain)	*_pattern (glob — MOTOR-*, SITE-?, etc.)

Transform quick settings

Node	Key options
flow_script_transform	language (JS/EQL), script, timeout_ms (default 500)
flow_change_originator	entity_type, id_field
flow_rename_keys	mapping (for example {"old":"new"})
flow_template	template (${data.x} / ${metadata.y} substitution)
flow_split	path (array location, default data)
flow_to_email	subject_template, body_template

Flow control quick settings

Node	Key options
flow_delay	delay_ms
flow_throttle	max_msgs, window_ms, (branches: SUCCESS/THROTTLED)
flow_debounce	window_ms
flow_merge	window_ms (accumulates in the data.merged array)
flow_subflow	target_flow_id
flow_retry	max_attempts (default 3), backoff_ms (default 1000), backoff_multiplier (default 2.0)
flow_log	level (INFO/WARN/ERROR), prefix
flow_noop	(no options)

External integration quick settings

Node	Static options	Dynamic options (*_field)
flow_http_request	method, url, headers, body_template, timeout_ms, retry_count, retry_delay_ms	url_field, method_field, body_field
flow_kafka_publish	bootstrap_servers, topic, value_template, headers	topic_field, key_field
flow_mqtt_publish	broker_url, topic, qos	topic_field
flow_webhook_callback	url, method, headers	url_field
flow_send_email	to, cc, subject, body	to_field, cc_field, subject_field, body_field
flow_send_sms	to, text	to_field, text_field
flow_send_push	title, body	title_field, body_field
flow_jdbc_query	dsn, sql, params_field	—

Actions — integration/storage quick settings

Node	Key options
flow_save_tag_point	tag_id_field (default metadata.tag_id), value_field, timestamp_field
flow_save_attributes	entity_type_field, id_field, attributes_field
flow_dds_publish	type, originator_field, payload_field
flow_publish_asset_event	asset_id_field, event_type_field, severity_field, details_field
flow_publish_asset_command	asset_id_field, tag_id, cmd_key_field, payload_field

Domain CRUD quick settings

Node	Key options
flow_create_*	Per-domain fields + the *_field dynamic option. Some NOT NULL fields are auto-filled (see the Domain CRUD table for the auto-filled items)
flow_update_*	Partial update: only the entered fields are updated, empty values are ignored. For a full overwrite, use Delete + Create
flow_delete_*	*_id_field
flow_start/end/pause/resume_work_order	order_id_field (default data.order_id)
flow_abort_work_order	+ abort_code_field, abort_notes_field
flow_update_tag_alarm_band_numeric	tag_id_field, hi_field, etc. (only the entered fields are updated)

Edge quick settings

All edge nodes use the same option set.

Option	Description
url	Edge REST endpoint (for example http://edge.local:60000/opc/server)
method	HTTP method (per-node default if unset)
headers	JSON headers (auth token, etc.)
body_template	Request body (if unset, data is sent as-is)
timeout_ms	5000

---

Flow REST API — Manipulating Flows Programmatically

A REST API for managing flows as code from external automation tools (Ansible/GitOps/CI pipelines), or for letting external systems execute a flow immediately.

Authentication

All API calls attach a token issued on the Security → API authentication tokens screen as a header.

Authorization: Bearer {api_token}

Endpoint list

1) List flows

GET /flow/list

curl -H "Authorization: Bearer ${TOKEN}" \
     https://platform.example.com/flow/list

Response:

{
  "data": [
    { "flow_id": "flow-abc123", "flow_name": "MES 동기화", "deployed": true,
      "node_count": 12, "exec_count": 9430, "error_count": 2, "last_exec_at": 1746247200000 },
    ...
  ]
}

2) Get a single flow

GET /flow/get/{flow_id}

The response body includes all graph nodes, relations, and options. It can be used directly for backup/version control.

3) Create a flow

POST /flow/create
Content-Type: application/json

{
  "flow_name":        "신규 자동화",
  "description":     "외부 알림 → 워크오더 자동 생성",
  "deployed":         false
}

Take flow_id from the response and use it in subsequent calls.

4) Update a flow (metadata)

POST /flow/update
Content-Type: application/json

{
  "flow_id":     "flow-abc123",
  "flow_name":   "수정된 이름",
  "description": "..."
}

5) Save the graph (bulk replace nodes and relations)

POST /flow/{flow_id}/graph
Content-Type: application/json

{
  "nodes":     [ { "node_id": "n1", "type": "flow_on_tag_point", "options": {...}, "x": 100, "y": 100 }, ... ],
  "relations": [ { "from_node_id": "n1", "to_node_id": "n2", "relation": "TRUE" }, ... ]
}

Saving the graph is transactional — if validation fails, the entire graph is rolled back.

6) Deploy / Undeploy

Changing the deployed field of the flow metadata deploys or undeploys it automatically.

# 배포
curl -X POST -H "Authorization: Bearer ${TOKEN}" -H "Content-Type: application/json" \
     -d '{"flow_id":"flow-abc123","deployed":true}' \
     https://platform.example.com/flow/update

7) Run immediately (manual trigger)

POST /flow/{flow_id}/run
Content-Type: application/json

{
  "type":       "WEBHOOK",
  "originator": { "entity_type": "External", "id": "manual-run" },
  "data":       { "test": true }
}

Executes immediately from the first node without passing through the trigger node's pre-filter. Useful for manual verification and debugging.

8) Dispatch (fire bypassing the trigger)

POST /flow/dispatch
Content-Type: application/json

{
  "type":       "POST_TELEMETRY",
  "originator": { "entity_type": "Tag", "id": "MOTOR-001.SPEED" },
  "data":       { "value": 1500 },
  "metadata":   { "tag_id": "MOTOR-001.SPEED", "ts": 1746247200000 }
}

Injects a message into the normal trigger processing path — every flow matching that message fires simultaneously.

9) Reset statistics

POST /flow/{flow_id}/stats/reset-errors        — 에러 카운트만 초기화
POST /flow/{flow_id}/stats/reset-all           — 전체 카운트 초기화

10) Export / Import — graph backup

# Export — JSON 파일로 다운로드
curl -H "Authorization: Bearer ${TOKEN}" \
     https://platform.example.com/flow/${FLOW_ID}/export \
     -o flow-backup.json

# Import — 같은 JSON 을 다른 환경에 등록
curl -X POST -H "Authorization: Bearer ${TOKEN}" -H "Content-Type: application/json" \
     --data @flow-backup.json \
     https://platform.example.com/flow/import

On import, if there is a conflicting ID, a new ID is issued automatically. Map external dependencies (credentials, asset IDs, etc.) separately after import.

11) Redeploy all flows

POST /flow/redeploy

Use it for bulk synchronization after large changes or after a node restart. Call it carefully during operation — it causes a temporary processing delay.

12) Self-test

POST /flow/selftest

Checks the flow engine's internal components (trigger queue, dispatcher, node registry, script runtime) once and returns their status. The result is also recorded in the diagnostic log.

13) Get the node catalog

GET /flow/catalog

Returns all currently registered node types and their option schemas. Used by the UI to generate the inspector dynamically. Also useful as a reference when an external tool generates graphs automatically.

Firing externally with a Webhook trigger

A flow containing a flow_on_webhook trigger can be fired directly by an external system at the following URL.

POST /flow/webhook/{flow_id}
Authorization: Bearer {token}    또는    X-API-Key: {token}
Content-Type:  application/json

{
  "order_no":  "PO-001",
  "customer":  "ACME",
  "quantity":  1000
}

Response:

{ "status": "ACCEPTED", "trace_id": "tr-..." }

Response code	Meaning
200	Enqueued into the message queue (the actual processing result is asynchronous)
401	Invalid or missing authentication token
404	Flow ID does not exist or is not deployed
413	Body exceeds 256 KB
422	The trigger node is not flow_on_webhook
429	Per-minute call limit exceeded

External automation tool integration examples

GitHub Actions — deploy a flow on PR merge

- name: 플로우 배포
  run: |
    curl -X POST \
      -H "Authorization: Bearer ${{ secrets.PP_TOKEN }}" \
      -H "Content-Type: application/json" \
      --data @flows/mes-sync.json \
      https://platform.example.com/flow/import

Ansible — batch flow management

- name: 모든 플로우 백업
  uri:
    url: "https://platform.example.com/flow/{{ item }}/export"
    headers:
      Authorization: "Bearer {{ pp_token }}"
    dest: "/backups/{{ item }}.json"
  loop: "{{ pp_flow_ids }}"

Jenkins — selftest on every new build

stage('PlantPulse Flow Selftest') {
  steps {
    sh """
      curl -X POST -H 'Authorization: Bearer ${PP_TOKEN}' \\
           https://platform.example.com/flow/selftest \\
           --fail-with-body
    """
  }
}

API rate limits

Endpoint	Per-minute limit (per token)
GET /flow/list · get · catalog	600
POST /flow/create · update · delete	60
POST /flow/{id}/run · dispatch · webhook/{id}	1,000
POST /flow/redeploy · selftest	10

Exceeding the limit returns 429 Too Many Requests.

---

OPC/PLC Industrial Integration Patterns

A collection of integration scenarios specific to industrial sites. Combines the 22 edge nodes with the 4 asset event publishing nodes.

Pattern A — Automatic OPC server registration

When a new line starts up, receive the line information from the ERP and automatically register an OPC server on the edge.

[flow_on_webhook]
   ↓ (data = {"line_id":"LINE-7","host":"10.0.7.10","port":4840})
[flow_change_originator]
   ↓ (originator → Edge:EDGE-A)
[flow_edge_opc_create]
   ↓ (path=/api/v1/opc, body={"opc_id":"OPC-${data.line_id}","host":"${data.host}","port":${data.port}})
[flow_edge_opc_start]
   ↓ (수집 시작 명령)
[flow_save_attributes]
   ↓ (등록 이력을 자산 메타에 기록)

Edge nodes automatically look up api_key from the mm_edge master table. Operators do not need to configure authentication separately.

Pattern B — Automatic PLC command publishing

Automatically issue a stop command to the PLC when an asset alarm occurs.

[flow_on_asset_alarm]
   ↓ where priority='ERROR' AND alarm_band='TRIP_HI'
[flow_publish_asset_command]
   ↓ (event_type='STOP', payload={"reason":"trip_high","triggered_by":"flow"})
[flow_log] (감사 로그)

flow_publish_asset_command is delivered immediately to the asset's command topic (asset/{id}/cmd/STOP), and the edge performs an OPC Write to the PLC.

Pattern C — Bulk OPC tag registration (CSV/Excel intake)

Register 100–1,000 tags from a CSV at once when setting up new equipment.

[flow_jdbc_poll]            ← 외부 DB 에 적재된 CSV 행
   ↓ (한 행 = 한 메시지)
[flow_script_transform]     ← 태그 정의 가공
   ↓
[flow_create_tag]           ← 도메인 등록 (NOT NULL 자동 보강)
   ↓
[flow_edge_tag_create]      ← 엣지에 동기화
   ↓
[flow_log] (성공 카운트)

When registering 1,000 entries, we recommend flattening to 100 or fewer per minute with flow_throttle. The edge may become temporarily slow to respond.

Pattern D — Automatic recovery from OPC disconnection

Restart automatically when an OPC connection drops and then recovers.

[flow_on_opc_status]
   ↓ where prev_status='CONNECTED' AND status='DISCONNECTED'
[flow_delay 30s]            ← 잠시 안정화 대기
   ↓
[flow_edge_opc_start]       ← 수집 재시작 시도
   ↓ SUCCESS: 정상
   └ FAILURE: [flow_retry max=3] → EXHAUSTED: [관리자 알림]

Pattern E — Line status check at shift start

Inspect all lines at a site at the start of each shift.

[flow_schedule cron='0 0 7,15,23 * * ?']   ← 7시·15시·23시
   ↓
[flow_edge_monitoring]      ← 엣지에서 OPC 상태 전체 조회
   ↓ data.opcs = [...]
[flow_split]                ← 배열을 행별 메시지로
   ↓
[flow_script_filter]        ← status != 'CONNECTED' 만
   ↓
[flow_send_email]           ← 비정상 OPC 만 한 통의 메일에 합쳐 발송

Pattern F — Automatic worker mapping on shift change

Automatically map workers to work orders according to roster changes.

[flow_on_entity_event type='ENTITY_UPDATED' Calendar]
   ↓
[flow_script_filter (시프트 시작 시각이 지금인 경우만)]
   ↓
[flow_jdbc_query (그 시프트의 작업자 목록 조회)]
   ↓ data.employees=[...]
[flow_split]
   ↓
[flow_update_work_order (작업자 매핑)]

Pattern G — Simultaneous calls to multiple PLCs (Scatter-Gather)

Call several PLCs simultaneously and consolidate the results.

[flow_on_webhook]
   ↓ data.targets=['PLC-1','PLC-2','PLC-3']
[flow_split]
   ↓ (3개로 분할)
[flow_edge_tag_read]
   ↓ (병렬 호출)
[flow_merge window=5s]
   ↓ data.merged=[...]
[flow_script_transform (data.summary 만들기)]
   ↓
[flow_publish_asset_aggregation]

Pattern H — Automatic alarm band adjustment

Automatically apply the learned LIMIT_MIN/MAX values from Predictive analytics to alarm bands.

[flow_schedule cron='0 0 4 * * MON']    ← 매주 월요일 4시
   ↓
[flow_jdbc_query (forecast 결과 조회)]
   ↓ data.tags=[{tag_id, limit_min, limit_max}]
[flow_split]
   ↓
[flow_update_tag_alarm_band_numeric]    ← lo=limit_min, hi=limit_max
   ↓ SUCCESS
[flow_log] (적용 결과 기록)

Pattern I — Automatic downtime classification

Classify asset stop events by reason so they are reflected accurately in OEE availability.

[flow_on_asset_event event_type='SHUTDOWN']
   ↓
[flow_switch]
   ├ CASE 점심: hour_of_day(ts) BETWEEN 12 AND 13 → [flow_create_calendar (계획정지)]
   ├ CASE 시프트 종료: 시프트 종료 시각 ±5분    → [flow_log (정상 종료)]
   ├ CASE 정기점검: 마지막 정비일 +30일 경과     → [flow_create_calendar (정기점검)]
   └ DEFAULT (비계획)                            → [flow_send_email (긴급)]

Pattern J — Bidirectional work order synchronization with an external ERP

PlantPulse ↔ external ERP bidirectional mirroring.

입력 1: ERP → 플랜트펄스
[flow_on_webhook] → [flow_create_work_order]

입력 2: 플랜트펄스 → ERP
[flow_on_entity_event type='ENTITY_CREATED' WorkOrder]
  → [flow_script_filter (출처가 ERP 가 아닌 경우만)]
  → [flow_http_request (ERP REST PUT)]

To prevent infinite loops, mark messages coming in from outside with metadata.source='ERP' and filter those messages out in the opposite-direction flow.

Cautions for industrial integration

Item	Recommendation
Issue OPC Write commands only after safety interlocks	Only after operator confirmation or passing an automatic safety rule
Do not set the PLC polling interval too short	Below 1 second risks a sharp rise in PLC CPU usage
Resource limits for Docker containers on edge devices	Be careful when OPC collection plus extra containers push CPU/memory above 80%
Use dry-run mode for all commands during commissioning	Use the dry_run=true option of flow_edge_tag_write
Prepare for power outages — trust only persistent triggers	flow_on_webhook and the like are volatile; domain event triggers are persistent

---

External System Integration Cookbook

A collection of flow_http_request settings and payload examples per frequently called external system.

Slack — Incoming Webhook notification

URL:    https://hooks.slack.com/services/T0000/B0000/{secret}
Method: POST
Headers:
  Content-Type: application/json
Body (body_template):
  {
    "text": "${data.title}",
    "blocks": [
      { "type": "header",  "text": { "type": "plain_text", "text": "${data.title}" } },
      { "type": "section", "text": { "type": "mrkdwn",     "text": "*자산:* ${metadata.asset_id}\n*값:* ${data.value}\n*시각:* ${metadata.ts}" } }
    ]
  }

• Success: 200 + body of ok
• Failure: 400 (bad payload) / 404 (bad secret) / 429 (per-minute call limit exceeded)
• Recommended: wire a flow_throttle for at most once per minute

Microsoft Teams — Webhook notification

URL:    https://{org}.webhook.office.com/webhookb2/{id}/IncomingWebhook/{secret}
Method: POST
Headers:
  Content-Type: application/json
Body:
  {
    "@type":    "MessageCard",
    "@context": "https://schema.org/extensions",
    "themeColor": "FF0000",
    "title": "${data.title}",
    "sections": [{
      "facts": [
        { "name": "자산", "value": "${metadata.asset_id}" },
        { "name": "값",   "value": "${data.value}"       },
        { "name": "시각", "value": "${metadata.ts}"      }
      ]
    }]
  }

Branch themeColor into FF0000 (red)/FFA500 (orange)/00C853 (green) to visualize priority.

Jira — Automatic issue creation

URL:    https://{org}.atlassian.net/rest/api/3/issue
Method: POST
Headers:
  Authorization: Basic {base64(email:apitoken)}
  Content-Type:  application/json
Body:
  {
    "fields": {
      "project":    { "key": "OPS" },
      "summary":    "[자동] ${data.title}",
      "description": {
        "type": "doc",
        "version": 1,
        "content": [{
          "type": "paragraph",
          "content": [{ "type": "text", "text": "자산: ${metadata.asset_id}\n값: ${data.value}" }]
        }]
      },
      "issuetype":  { "name": "Bug" },
      "priority":   { "name": "High" }
    }
  }

• An issue key such as OPS-1234 arrives in the response body data.response_body.key, so use it in downstream nodes
• Failures: 400 (bad field) / 401 (authentication) / 403 (no project permission)

GitHub Issues — Automatic issue creation

URL:    https://api.github.com/repos/{owner}/{repo}/issues
Method: POST
Headers:
  Authorization: Bearer {pat_token}
  Accept:        application/vnd.github+json
Body:
  {
    "title":  "[자동] ${data.title}",
    "body":   "자산: ${metadata.asset_id}\n값: ${data.value}\n시각: ${metadata.ts}\ntrace: ${metadata.trace_id}",
    "labels": ["automation", "ops"]
  }

ERP (SAP S/4HANA OData) — Work order issuance

URL:    https://{host}/sap/opu/odata/sap/API_MAINTNOTIFICATION/MaintenanceNotification
Method: POST
Headers:
  Authorization:   Basic {base64(user:pass)}
  X-CSRF-Token:    fetch          ← 별도 GET 으로 토큰 받기
  Content-Type:    application/json
Body:
  {
    "NotificationType":         "M2",
    "MaintenanceNotificationType": "M2",
    "TechnicalObject":          "${metadata.asset_id}",
    "NotificationText":         "${data.title}",
    "MalfunctionStartDate":     "${data.start_date}",
    "Priority":                 "${data.priority}"
  }

You must first fetch the CSRF token with a GET and then POST with the same session cookie. Connect two flow_http_request nodes and propagate the token from the first node's response headers into the second node's headers.

MES (direct INSERT into an external DB) — flow_jdbc_query

Insert rows directly into an external MES system's work order table.

INSERT INTO mes.work_order
  (order_no, asset_id, product_id, qty, status, due_date, created_by)
VALUES
  (:order_no, :asset_id, :product_id, :qty, 'NEW', :due_date, 'flow')

Node option	Value
datasource_id	mes_db (registered in advance in the system settings)
query	The SQL above
binds	{"order_no":"${data.order_no}","asset_id":"${metadata.asset_id}",...}

flow_jdbc_query supports INSERT/UPDATE/DELETE in addition to SELECT. Transactions are automatically committed/rolled back per node.

Grafana — Dashboard auto-refresh notification

URL:    https://{host}/api/annotations
Method: POST
Headers:
  Authorization: Bearer {api_key}
  Content-Type:  application/json
Body:
  {
    "dashboardUID": "...",
    "panelId":      4,
    "time":         ${metadata.ts},
    "tags":         ["alarm", "${metadata.asset_id}"],
    "text":         "${data.title}"
  }

A marker is automatically displayed on the graph when an alarm occurs — extremely useful for post-mortem analysis.

Telegram — Bot message

URL:    https://api.telegram.org/bot{token}/sendMessage
Method: POST
Body:
  {
    "chat_id": "-1001234567890",
    "text":    "🚨 *${data.title}*\n자산: `${metadata.asset_id}`\n값: ${data.value}",
    "parse_mode": "Markdown"
  }

Quick table of REST authentication patterns

Authentication method	Header
API Key (header)	X-API-Key: {token}
API Key (query)	?api_key={token} at the end of the URL
Bearer Token	Authorization: Bearer {token}
Basic Auth	Authorization: Basic {base64(user:pass)}
OAuth2 (Client Credentials)	Obtain a token → Bearer header (refreshed by a separate node)
HMAC signature	Compute a body/time-based signature in a script and put it in the X-Signature header

Response parsing patterns

After a flow_http_request response, use data.response_body in the next node:

// 응답에서 ID 추출
data.created_id = data.response_body.id;
data.status     = data.response_body.status;

// 응답 배열에서 첫 행만
data.first_item = (data.response_body.items || [])[0] || null;

// 응답이 문자열이면 JSON 파싱
if (typeof data.response_body === 'string') {
  try { data.response_body = JSON.parse(data.response_body); } catch (e) {}
}
msg

Security recommendations for external system calls

Item	Recommendation
Do not leave tokens in plaintext inside the graph	Register them in System settings → credential store and reference them like ${creds.slack_webhook}
Mask sensitive information in response bodies	Remove PII/tokens with flow_script_transform before the next node
A separate retry policy per external system	For high call volumes, wire a dedicated flow_retry + flow_throttle
Store call results as audit logs	Add call history metadata to the asset with flow_save_attributes

---

Data Transformation Cookbook

A collection of transformation patterns you can copy straight into the flow_script_transform node. All examples are JavaScript and use the data / metadata shorthand aliases.

Unit conversion

// 섭씨 → 화씨
data.temp_f = data.temp_c * 9 / 5 + 32;

// 바 → kPa
data.pressure_kpa = data.pressure_bar * 100;

// rpm → rad/s
data.angular_velocity = data.rpm * 2 * Math.PI / 60;

// kWh → MJ
data.energy_mj = data.energy_kwh * 3.6;

// 바이트 → MB (소수 1자리)
data.size_mb = Math.round(data.bytes / 1024 / 1024 * 10) / 10;

msg

Date and time conversion

const d = new Date(metadata.ts);

// ISO 8601 — 2026-05-12T12:34:56.789Z
data.iso = d.toISOString();

// 사람용 — 2026-05-12 21:34:56 (KST)
data.local = d.toLocaleString('ko-KR', { hour12: false });

// 날짜만 — 2026-05-12
data.date = d.toISOString().slice(0, 10);

// 시간만 — 21:34:56
data.time = d.toTimeString().slice(0, 8);

// 분 단위로 내림 (스파크라인 키)
data.minute_key = Math.floor(metadata.ts / 60000) * 60000;

// 시간 단위로 내림
data.hour_key = Math.floor(metadata.ts / 3600000) * 3600000;

// 한국 시간 (UTC+9) 직접 더하기 (서버 시간이 UTC 인 경우)
data.kst_hour = new Date(metadata.ts + 9 * 3600000).getUTCHours();

msg

String normalization

// 공백 trim + 소문자화
data.normalized = (data.text || '').trim().toLowerCase();

// 한글·영문 외 제거 (특수문자/공백 정리)
data.clean = (data.text || '').replace(/[^가-힣a-zA-Z0-9]/g, '');

// camelCase → snake_case
data.snake = (data.text || '').replace(/([A-Z])/g, '_$1').toLowerCase().replace(/^_/, '');

// 전화번호 정규화 (숫자만)
data.phone_digits = (data.phone || '').replace(/\D/g, '');

// 한국 휴대전화 자동 포맷 (010-1234-5678)
const p = (data.phone || '').replace(/\D/g, '');
data.phone_formatted = p.length === 11 ? p.replace(/(\d{3})(\d{4})(\d{4})/, '$1-$2-$3') : p;

msg

Flattening nested objects

// data.location.address.city → data.city
function flatten(obj, prefix, out) {
  out = out || {};
  for (var k in obj) {
    var v = obj[k];
    var key = prefix ? prefix + '_' + k : k;
    if (v && typeof v === 'object' && !Array.isArray(v)) flatten(v, key, out);
    else out[key] = v;
  }
  return out;
}
data = flatten(data);
msg

Flat object → nested

// data.user_name + data.user_email → data.user = {...}
function unflatten(obj) {
  var out = {};
  for (var k in obj) {
    var parts = k.split('_');
    var cur = out;
    for (var i = 0; i < parts.length - 1; i++) {
      cur[parts[i]] = cur[parts[i]] || {};
      cur = cur[parts[i]];
    }
    cur[parts[parts.length - 1]] = obj[k];
  }
  return out;
}
data = unflatten(data);
msg

Generating CSV rows

// 외부 시스템에 CSV 한 줄 보낼 때
function csvEscape(v) {
  if (v === null || v === undefined) return '';
  v = String(v);
  return /[,"\n]/.test(v) ? '"' + v.replace(/"/g, '""') + '"' : v;
}
data.csv_row = [
  csvEscape(metadata.ts),
  csvEscape(metadata.asset_id),
  csvEscape(data.value),
  csvEscape(data.quality)
].join(',');
msg

Array aggregation

var arr = data.values || [];

// 합·평균·최소·최대
data.sum = arr.reduce(function(a, b) { return a + b; }, 0);
data.avg = arr.length ? data.sum / arr.length : 0;
data.min = arr.length ? Math.min.apply(null, arr) : null;
data.max = arr.length ? Math.max.apply(null, arr) : null;

// 중앙값
var sorted = arr.slice().sort(function(a, b) { return a - b; });
var mid    = Math.floor(sorted.length / 2);
data.median = sorted.length === 0 ? null
            : sorted.length % 2 ? sorted[mid]
            : (sorted[mid - 1] + sorted[mid]) / 2;

msg

Conditional field addition (schema evolution)

// 알람 우선순위에 따라 색상·아이콘 자동 부여
var p = data.priority || 'INFO';
data.color   = { ERROR: '#d32f2f', WARN: '#f57c00', INFO: '#1976d2' }[p] || '#9e9e9e';
data.icon    = { ERROR: '🔴',       WARN: '🟠',       INFO: '🔵' }[p] || '⚪';
data.urgency = p === 'ERROR' ? 3 : p === 'WARN' ? 2 : 1;
msg

Masking part of a payload

function maskEmail(e) {
  if (!e || e.indexOf('@') < 0) return e;
  var parts = e.split('@');
  return parts[0].slice(0, 2) + '***@' + parts[1];
}
function maskPhone(p) {
  return (p || '').replace(/(\d{3})\d{4}(\d{4})/, '$1-****-$2');
}
data.user_email = maskEmail(data.user_email);
data.user_phone = maskPhone(data.user_phone);
msg

Multilingual (i18n) templates

// 메시지 본문을 사용자 로케일에 따라 분기
const tpl = {
  'ko-KR': '🚨 ${asset} 온도 ${value}°C 임계 초과',
  'en-US': '🚨 ${asset} temperature ${value}°C exceeds threshold',
  'ja-JP': '🚨 ${asset} 温度 ${value}°C 閾値超過'
};
const locale = metadata.user_locale || 'ko-KR';
const template = tpl[locale] || tpl['ko-KR'];
data.notification = template
  .replace('${asset}', metadata.asset_id)
  .replace('${value}', data.value);
msg

Safe JSON Path lookup

// data.deep.nested.field 처럼 깊은 경로를 안전하게 조회
function get(obj, path, dflt) {
  var keys = path.split('.');
  var cur  = obj;
  for (var i = 0; i < keys.length; i++) {
    if (cur === null || cur === undefined) return dflt;
    cur = cur[keys[i]];
  }
  return cur === undefined ? dflt : cur;
}
data.city = get(data, 'location.address.city', 'Unknown');
msg

Message merging (post-processing flow_merge)

// flow_merge 가 data.merged 배열로 모은 메시지들을 1건으로 합침
var items = data.merged || [];
data.summary = {
  count:     items.length,
  first_ts:  items[0]?.metadata?.ts,
  last_ts:   items[items.length - 1]?.metadata?.ts,
  assets:    Array.from(new Set(items.map(function(m) { return m.metadata?.asset_id; }))),
  max_value: Math.max.apply(null, items.map(function(m) { return m.data?.value || 0 }))
};
delete data.merged;
msg

---

Reusable Script Collection

A script library you can copy straight into the flow_script_filter / flow_script_transform / flow_switch nodes. All scripts are JavaScript and use the data / metadata shorthand aliases.

Transform scripts

Unit conversion and labeling

// 섭씨 → 화씨 + 등급 라벨
data.temp_f = data.temp * 1.8 + 32;
data.grade  = data.temp > 80 ? 'HIGH' : data.temp < 0 ? 'LOW' : 'OK';
msg

Enriching time, shift, and weekday metadata

const d = new Date(metadata.ts);
const h = d.getHours();
metadata.shift   = (h >= 6 && h < 14) ? 'A' : (h < 22) ? 'B' : 'C';
metadata.weekday = ['SUN','MON','TUE','WED','THU','FRI','SAT'][d.getDay()];
metadata.is_weekend = (d.getDay() === 0 || d.getDay() === 6);
msg

Computing an average ± 3σ alarm band

const m = data.mean, s = data.stddev || 1;
data.tag_id = originator.id + '.TEMP';
data.hi     = m + 3*s;  data.lo    = m - 3*s;
data.hi_hi  = m + 4*s;  data.lo_lo = m - 4*s;
data.use_alarm = true;
msg

Mapping an MES PO → work order

const po = data;
data = {
  master_id: 'WO-MES-' + po.po_no,
  asset_id:  po.line_id || 'UNASSIGNED',
  title:     po.product_name + ' (' + po.qty + ')',
  due_date:  po.delivery_date,
  qty:       po.qty,
  product_id: po.product_code
};
msg

Normalizing external payloads (unifying various field names)

// 외부 시스템에 따라 키 이름이 다른 경우 — 한 줄로 정규화
data.value     = data.value ?? data.val ?? data.v;
data.timestamp = data.timestamp ?? data.ts ?? metadata.ts;
data.tag_id    = data.tag_id ?? data.tagId ?? data.id;
msg

Accumulating threshold violation counts (stateful pattern)

// 자산 attribute 와 함께 사용 — 상태는 메시지 자체에 적재
data.consecutive_fail = (data.consecutive_fail ?? 0) + (data.passed ? 0 : 1);
data.alert            = data.consecutive_fail >= 5;
msg

Message statistics summary (processing Merge results)

// flow_merge 후 data.merged 배열을 받아 요약
const arr = data.merged || [];
data.count   = arr.length;
data.values  = arr.map(m => m.data.value).filter(v => v != null);
data.mean    = data.values.reduce((a,b)=>a+b, 0) / (data.values.length || 1);
data.max     = Math.max(...data.values);
data.min     = Math.min(...data.values);
delete data.merged;
msg

Applying time-of-day thresholds

const h = new Date(metadata.ts).getHours();
const threshold = (h >= 8 && h < 18) ? 90 : 70;   // 주간 90, 야간 70
data.alarm = data.value > threshold;
data.threshold = threshold;
msg

Filter scripts

Priority whitelist

['ERROR', 'CRITICAL'].includes(data.priority)

Time window (daytime only)

const h = new Date(metadata.ts).getHours();
h >= 8 && h < 20

Asset/tag ID patterns

/^MOTOR-.*$/.test(originator.id)
// 또는
metadata.tag_id && metadata.tag_id.startsWith('LINE-A.')

Threshold + stability (N consecutive times)

data.value > 100 && (data.consecutive_count ?? 0) >= 5

Business hours and holiday check

const d = new Date(metadata.ts);
const h = d.getHours();
const wd = d.getDay();
// 평일 09-18시만 통과
wd >= 1 && wd <= 5 && h >= 9 && h < 18

Specific site only

['SITE-01', 'SITE-02'].includes(metadata.site_id)

Checking that all required fields exist

data.value != null && data.tag_id && metadata.ts

Blocking self-published messages (bidirectional synchronization)

metadata.source !== 'flow'

Switch case scripts

By priority grade

// case "Critical"
['CRITICAL', 'EMERGENCY'].includes(data.priority)

// case "High"
data.priority === 'ERROR'

// case "Normal"
['WARN', 'INFO'].includes(data.priority)

By asset state

// case "Running"
data.status === 'RUN'

// case "Stopped"
['STOP', 'IDLE', 'PAUSED'].includes(data.status)

// case "Faulted"
data.status === 'FAULT' || data.error_count > 0

By work order lifecycle

// case "Start"
data.event_type === 'START_REQUEST'

// case "End"
data.event_type === 'COMPLETE' && data.qty_done >= data.qty_planned

// case "Abort"
data.event_type === 'CANCEL'

All of the scripts above collect patterns frequently used in production environments. They work when wired directly into a graph; just adjust the thresholds and field names to match your domain.

---

Glossary

Flow engine terms

Term	Meaning
entity_type	The kind of subject entity of a message — Asset, Tag, Site, Order, Customer, Product, Employee, Calendar, etc.
originator	The subject entity the message refers to (entity_type + id). For example Asset/MOTOR-001
type	The message classification label. Identifies which kind of event the trigger received
data	The message body (payload) — mainly read and written by transform and action nodes
metadata	The message context (time, site, shift, tag ID, etc.) — preserved through to the end of the flow even after transformations
relation	The label on a node's output wire. SUCCESS/FAILURE/TRUE/FALSE/MATCH/NO_MATCH/DEFAULT/THROTTLED/EXHAUSTED (all uppercase)
*_field dynamic option	An input method that reads a value from a path in the message payload (for example data.tag_id) instead of a static value
Glob pattern	Wildcard notation for the trigger *_pattern option — * matches zero or more characters, ? matches exactly one
Snapshot	A per-version backup stored automatically when the graph is saved. Used to revert to a previous state after an incident
Partial update (fetch+merge)	The way an Update node fetches the existing record and merges only the fields you entered. Empty values are ignored
SKIPPED	Handling in which a message that does not match the trigger pattern does not flow to downstream nodes and is excluded from counters
EXHAUSTED	The branch fired when a flow_retry node exceeds the maximum number of retries
THROTTLED	The branch fired when a flow_throttle node blocks a message that exceeded the limit within the window

Domain and abbreviation glossary

Abbreviations commonly used in plant operations, organized so you can look them up quickly within this manual.

Abbreviation	Meaning	In this manual
MES	Manufacturing Execution System — work order and production performance management	Target of external integration (HTTP/MQTT/external DB)
ERP	Enterprise Resource Planning — enterprise-wide resource and planning system	Target of external integration
SCADA	Supervisory Control and Data Acquisition	Target of external integration / exit point of flow_publish_asset_command
OPC	Open Platform Communications — industrial communication standard	Target of the Edge category (flow_edge_opc_*)
OEE	Overall Equipment Effectiveness — availability × performance × quality	flow_on_oee_event trigger
RAM	Reliability, Availability, Maintainability	flow_on_ram_event trigger
EMS	Energy Management System	flow_on_ems_event trigger
EQL	Domain event rule expression language	The language: "EQL" option of flow_script_filter/flow_script_transform
CEP	Complex Event Processing	The EQL-based rule engine. Alarms must be raised only through the CEP path to remain consistent
PO	Purchase Order — purchase/production order	The unit converted into a work order during MES integration
WO	Work Order	The flow_*_work_order node family
CMMS	Computerized Maintenance Management System	Target of external integration
MTTF/MTTR	Mean Time To Failure / To Repair	Key indicators of RAM events
HMI	Human–Machine Interface	Operating screens such as SCADA

---

Version Notes

When the major feature groups covered by this manual were introduced. If you are running an older version, some features may behave differently.

V2026.05 — Domain automation expansion

• 11 new Edge category nodes — OPC server/tag CRUD + monitoring automation
• 5 work order state transitions — flow_start/end/pause/resume/abort_work_order
• 2 tag alarm band partial updates — updates only the entered fields of numeric/boolean alarm bands
• New flow_retry flow control — back-off + maximum attempts + EXHAUSTED branch
• 5 new triggers — flow_on_asset_health_status / _connection_status / _oee_event / _ram_event / _ems_event
• Update node partial update (fetch+merge) — fetches the existing record and merges only the entered fields
• Relations standardized to uppercase — SUCCESS/FAILURE/... all uppercase; existing graphs are converted automatically
• SKIPPED handling — excludes trigger-pattern non-matching messages from counters
• Redeploy All operational tool — bulk reload of active flows
• Reset all counts — bulk reset of node- and flow-level statistics

V2026.03 — Release stability

• Automatic snapshots stored on graph save
• Live debug panel (2-second refresh at the bottom of the inspector) + node indicators and duration display
• import/export (graph JSON files)
• Load watchdog introduced — emits a one-line diagnostic log when load persists

Earlier

• M1 — engine and UI skeleton, graph CRUD, visual canvas
• M2 — domain action nodes (asset, tag, work order, production domains, and so on)
• M3 — trigger diversification, 8 filter/transform/external integration nodes, script engine
• M4 — debug storage and execution history screen, statistics, deploy/undeploy, import/export
• M5 — domain trigger integration, bulk scenario verification

All features and behavior covered in this manual are based on V2026.05 above.

---

Related Screens

• CEP (Complex Event Processing): the EQL-based domain event rule engine
• Alarms: alarm occurrence and history
• Data Points: querying and analyzing tag data
• Developer guide: Flow Engine — engine architecture, node interfaces, DB schema