node-red-contrib-edgelink-pg 1.2.1

EdgeLink PostgreSQL/TimescaleDB 批量写入节点 — 三菱PLC/Modbus 采集数据直写入库，支持断线缓存重试

npm install node-red-contrib-edgelink-pg

node-red-contrib-edgelink-pg

PostgreSQL/TimescaleDB batch-write node for industrial SCADA edge computing. Zero-code pipeline: PLC data → database, with buffer, retry, and connection pooling.

Why This Node?

Installation

cd ~/.node-red
npm install node-red-contrib-edgelink-pg

Requires Node.js ≥ 14 and Node-RED ≥ 3.0. The pg driver is installed automatically.

Nodes

edgelink-pg-config (config)

Stores PostgreSQL connection parameters. Multiple store nodes can reference a single config — they share one connection pool.

edgelink-pg-store (output)

Receives data, buffers, batch-inserts, retries on failure.

Input Formats (Auto-Detected)

The node automatically detects which format an incoming message uses. No configuration required.

Format 1 — MC Driver (mitsubishi-read / modbus-read)

msg.payload = {
  success: true,
  deviceId: "PLC-Oven-01",
  data: {
    "Temperature": {
      rawValue: 2530,          // raw PLC value
      engValue: 253.0,         // engineering value
      quality: 0,              // 0 = good
      ts: "2026-06-27T16:00:00.000Z",
      regType: "D"             // PLC register type
    },
    "Switch": {
      rawValue: 1, engValue: 1,
      quality: 0, ts: "2026-06-27T16:00:00.000Z",
      regType: "X"
    }
  }
}

Detection: payload.data exists with nested tag objects.
Table schema: fixed 7 columns (insert_time, device_id, tag_id, register_type, raw_value, eng_value, quality).
Supports ON CONFLICT DO NOTHING, autoCreateTable, TimescaleDB hypertable, and ${deviceId} dynamic sharding.

Format 2 — Batch Rows

msg.payload = {
  rows: [
    { sensor: "temp", value: 25.5, ts: "2026-06-29T16:00:00Z" },
    { sensor: "press", value: 1.2, ts: "2026-06-29T16:00:00Z" }
  ]
}
msg.topic = "sensor_data"   // table name (optional)

Detection: payload.rows is an array.
Columns: auto-detected from Object.keys() of the first row.
Table name: msg.tableName → msg.topic → node config tableName.

Format 3 — Single-Row Object

msg.payload = {
  sensor: "temp",
  value: 25.5,
  ts: "2026-06-29T16:00:00Z"
}
msg.topic = "sensor_data"

Detection: plain object (no .data, no .rows).
Auto-wrapped as a single-row batch.

Dynamic Controls (All Formats)

Output Format

消息仅在 flush 完成 或 写入失败 时发出（v1.2.0+），缓冲期间静默。通过 event 字段区分：

// 写入成功
{ event: "flushed", success: true,  inserted: 100, failed: 0,
  buffered: 0, retryBuffered: 0, tableName: "plc_data", roundTimeMs: 45 }

// 写入失败（进入重试队列）
{ event: "retry",   success: false, inserted: 0,   failed: 100,
  buffered: 0, retryBuffered: 100, error: "[CONNECTION] ...", tableName: "plc_data" }

// 数据错误（丢弃）
{ event: "error",   success: false, inserted: 0,   failed: 50,
  buffered: 0, retryBuffered: 0,  error: "[DATA_ERROR] ...", tableName: "plc_data" }

Architecture

 ┌──────────┐    ┌─────────────────────────────┐    ┌──────────┐
 │ mitsubishi│    │     edgelink-pg-store        │    │PostgreSQL│
 │  -read    │───→│                              │───→│/Timescale│
 │   PLC     │    │  buffer[] ──→ batch INSERT   │    │    DB    │
 └──────────┘    │    │           (parameterized) │    └──────────┘
                 │    ├──→ success → output       │
 ┌──────────┐    │    └──→ failure → retryBuffer  │
 │ inject   │───→│         │                      │
 │(generic) │    │         └──→ retryInterval     │
 └──────────┘    │              (fixed, no backoff)│
                 └─────────────────────────────┘

 Pool: { "user@host:port/db" → pg.Pool }  (global singleton, ref-counted, connectTimeout 5s)
 Mutex: _writing  (prevents concurrent INSERT)
 Schema: _pendingSchemas[]  (queued during _writing, never lost)
 TTL:  _tableCreationFailed (5-min expiry, auto-retry after manual table creation)

Key behaviors (v1.2.0):

• Flush triggers: buffer.length ≥ batchSize OR msg.flush === true OR flushInterval timer
• _writing lock: Only one INSERT in flight at a time; concurrent triggers are silently skipped
• Column/table change: Queued via _pendingSchemas[] during flush, applied after completion
• retryBuffer: FIFO queue; capped by total row count (retryRowMax); event-driven timer
• Output: Only emits on actual flush/retry (not per-input); uses event field to distinguish
• Table create: Failed auto-create TTL expired after 5 min (allows manual fix + retry)

Error Classification

Table Structure (MC Format)

When autoCreateTable = true:

CREATE TABLE IF NOT EXISTS plc_data (
    insert_time   TIMESTAMPTZ NOT NULL,
    device_id     VARCHAR(64) NOT NULL,
    tag_id        VARCHAR(64) NOT NULL,
    register_type VARCHAR(8),
    raw_value     NUMERIC,
    eng_value     NUMERIC,
    quality       INTEGER DEFAULT 0,
    PRIMARY KEY (insert_time, device_id, tag_id)
);

CREATE INDEX IF NOT EXISTS idx_plc_data_dt
    ON plc_data (device_id, tag_id, insert_time DESC);

With useTimescaleDB = true:

SELECT create_hypertable('plc_data', 'insert_time', if_not_exists => TRUE);

Primary key limitation: Sub-second duplicate (time, device, tag) records are rejected via ON CONFLICT DO NOTHING. Recommended polling interval ≥ 1 second.

Quick Start

1. Add config node

Drag edgelink-pg-config to the canvas. Fill in your PostgreSQL credentials.

2. Wire up

[mitsubishi-read] → [edgelink-pg-store] → [debug]
                        ↑
               [edgelink-pg-config]

3. Deploy

Data starts flowing. Status indicator: 🟢 green (inserting) / 🟡 yellow (buffering) / 🔴 red (error).

4. Standalone test (no PLC needed)

Use an inject node with JSON payload:

{"sensor": "temp", "value": 25.5}

Set msg.topic to sensor_data, wire to edgelink-pg-store. Deploy and press the inject button.

Integration with EdgeLink Ecosystem

┌──────────────┐     ┌──────────────────┐     ┌──────────────────────┐
│ mitsubishi-  │────→│ edgelink-pg-store│────→│ PostgreSQL/Timescale │
│ read (3E/4E) │     │                  │     │                      │
└──────────────┘     │  • auto-detect   │     │  • plc_data          │
                     │  • batch INSERT  │     │  • hypertable (opt)  │
┌──────────────┐     │  • retry on fail │     │                      │
│ modbus-read  │────→│  • ON CONFLICT   │     └──────────────────────┘
└──────────────┘     └──────────────────┘

Compatible with any upstream node that outputs the standard EdgeLink format.

Status Indicator

FAQ

Q: Can I execute arbitrary SQL?
No. This is a data logger, not a SQL client. Use node-red-contrib-postgresql for custom queries.

Q: Does it support table auto-creation for generic formats?
No. Auto-create is MC format only (fixed schema). For generic formats, create the table manually.

Q: What happens if data arrives while a flush is in progress?
Data is appended to the buffer. The next flush trigger (batch size, timer, or msg.flush) will pick it up. No data is lost.

Q: How is the connection pool shared across nodes?
A module-level POOLS map, keyed by user@host:port/database. Reference-counted. Last node standing calls pool.end().

Q: Why no exponential backoff for retries?
Edge scenarios: PG either recovers quickly (network hiccup) or stays dead (needs human). Fixed 5s retry is sufficient; backoff adds state without benefit.

License

MIT — see LICENSE for full text.