Unified Backend API Documentation

The unified-backend is a high-performance Rust API server built on Actix-web that provides REST and WebSocket endpoints for real-time power quality monitoring. It serves as the central data gateway for the EQ Synapse platform.

Architecture Overview

Technology Stack

Web Framework: Actix-web 4.11.0 (async, actor-based)
Database: DuckDB (persistent Arrow-optimized storage)
Data Format: Apache Arrow IPC (zero-copy, columnar)
Serialization: Arrow IPC (Inter-Process Communication)
WebSocket: Actix-web-actors with custom session management
Shared Memory: ArrowSharedMemory for live streaming

Server Configuration

Host: 0.0.0.0
Port: 8080
Workers: 2 Actix HTTP workers
Worker Pool: 1 data processing worker (configurable)
Keep-Alive: 75 seconds

Middleware Stack

CORS: Allow any origin, method, header (supports credentials)
Logger: Request/response logging
Compress: Response compression (gzip, deflate, brotli)

API Base Path

All endpoints are prefixed with /api/v1

Core Concepts

Apache Arrow IPC Format

All HTTP data responses use Apache Arrow IPC (Inter-Process Communication) format:

Content-Type: application/vnd.apache.arrow.stream

Benefits:

Zero-copy deserialization
Language-agnostic (Python, JavaScript, Rust, C++)
Columnar format (efficient for analytics)
Preserves schema and data types

Client Libraries:

JavaScript: apache-arrow npm package
Python: pyarrow package
Rust: arrow crate

Worker Pool Architecture

Requests are processed by a worker pool with:

LRU Cache: 60-second TTL, 1000 entries
Query Timeout: 30 seconds
Data Strategies:
- Parquet: Historical data (>24h old)
- InMemory: Live data (<24h old)
- Hybrid: Queries spanning both timeframes

Data Sources

DuckDB (Persistent): Parquet-backed tables for historical data
Shared Memory (Live): ArrowSharedMemory for real-time CPOW streaming
Event Logs: JSON files in events/{device_id}/ directories

REST API Endpoints

Health & Status

`GET /health`

Simple health check endpoint.

Response:

200 OK
"Server is healthy"

CPOW (Continuous Point-on-Wave) Endpoints

`GET /api/v1/devices/{id}/cpow/data`

Query historical CPOW waveform data.

Path Parameters:

id (string): Device identifier (e.g., “wave001”)

Query Parameters:

start_time (optional): ISO 8601 timestamp in UTC
end_time (optional): ISO 8601 timestamp in UTC
metrics (optional): Comma-separated metric names
limit (optional): Maximum number of rows to return

Response:

Content-Type: application/vnd.apache.arrow.stream
Body: Arrow IPC binary stream

Example Request:

curl "http://localhost:8080/api/v1/devices/wave001/cpow/data?start_time=2026-01-01T00:00:00Z&limit=10000" \
  -H "Accept: application/vnd.apache.arrow.stream"

Python Example:

import requests
import pyarrow as pa

response = requests.get(
    "http://localhost:8080/api/v1/devices/wave001/cpow/data",
    params={"start_time": "2026-01-01T00:00:00Z", "limit": 10000}
)
table = pa.ipc.open_stream(response.content).read_all()
df = table.to_pandas()
print(df.head())

Error Responses:

404: Device not found or no data available
400: Invalid query parameters
408: Query timeout (>30 seconds)
500: Internal server error

`GET /api/v1/devices/{id}/cpow_data_latest`

Get the most recent CPOW data for a device.

Response:

Content-Type: application/vnd.apache.arrow.stream
Body: Arrow IPC binary with latest waveform data

`GET /api/v1/devices/{id}/cpow_data_directional`

Query CPOW data directionally from a specific timestamp.

Query Parameters:

from_timestamp (required): Unix microseconds
direction (required): “ascending” or “descending”
limit (optional): Row limit (default: 640,000)

Example:

curl "http://localhost:8080/api/v1/devices/wave001/cpow_data_directional?from_timestamp=1735689600000000&direction=descending&limit=100000"

`GET /api/v1/devices/cpow/list_files`

List available CPOW parquet files.

Response:

{
  "files": [
    "cpow_2026_01_01.parquet",
    "cpow_2026_01_02.parquet"
  ]
}

Note: Excludes the most recent file (may be actively written to).

`POST /api/v1/devices/cpow/plot_with_file_name`

Load CPOW data from a specific parquet file.

Request Body:

{
  "file_name": "cpow_2026_01_01.parquet"
}

Response:

Content-Type: application/vnd.apache.arrow.stream

Security: Path traversal protection (rejects .., /, \ in filenames).

PMON (Power Monitoring) Endpoints

`GET /api/v1/devices/{id}/pmon/data`

Query historical PMON aggregate data.

Query Parameters:

start_time (optional): ISO 8601 timestamp in UTC
end_time (optional): ISO 8601 timestamp in UTC
metrics (optional): Comma-separated field names
limit (optional): Maximum rows to return

Available Metrics:

Category	Metrics
Voltage (RMS)	`AVRMS`, `BVRMS`, `CVRMS`
Current (RMS)	`AIRMS`, `BIRMS`, `CIRMS`, `NIRMS`
Real Power	`AWATT`, `BWATT`, `CWATT`
Fundamental Voltage	`AFVRMS`, `BFVRMS`, `CFVRMS`
Fundamental Current	`AFIRMS`, `BFIRMS`, `CFIRMS`
Fundamental Power	`AFWATT`, `BFWATT`, `CFWATT`
Reactive Power	`AFVAR`, `BFVAR`, `CFVAR`
Frequency	`FREQ`

Example:

import requests
import pyarrow as pa

response = requests.get(
    "http://localhost:8080/api/v1/devices/wave001/pmon/data",
    params={
        "start_time": "2026-01-01T00:00:00Z",
        "end_time": "2026-01-01T01:00:00Z",
        "metrics": "AVRMS,FREQ",
        "limit": 10000
    }
)
table = pa.ipc.open_stream(response.content).read_all()
df = table.to_pandas()

`POST /api/v1/devices/{id}/thumbnail`

Get thumbnail/preview data for a single PMON field.

Request Body:

{
  "field": "AVRMS",
  "limit": 1000,
  "timeRange": 3600,
  "startTime": "2026-01-01T00:00:00Z",
  "endTime": "2026-01-01T01:00:00Z"
}

Parameters:

field (required): Single PMON field name (e.g., “AVRMS”, “FREQ”)
limit (optional): Row limit (default: 1000)
timeRange (optional): Seconds from now (ignored if start/end provided)
startTime (optional): ISO 8601 timestamp
endTime (optional): ISO 8601 timestamp

Response:

Content-Type: application/vnd.apache.arrow.stream

`GET /api/v1/devices/pmon/list_files`

List available PMON parquet files.

Response:

{
  "files": [
    "pmon_2026_01_01.parquet",
    "pmon_2026_01_02.parquet"
  ]
}

`POST /api/v1/devices/pmon/plot_with_file_name`

Load PMON data from a specific parquet file.

Request Body:

{
  "file_name": "pmon_2026_01_01.parquet"
}

Response:

Content-Type: application/vnd.apache.arrow.stream

Power Quality Events

`POST /api/v1/events/ingest`

Ingest a new power quality event.

Request Body:

{
  "event_id": "evt_123",
  "event_type": "voltage_sag",
  "timestamp_start": 1735689600000000,
  "timestamp_end": 1735689650000000,
  "timestamp_formatted": "2026-01-01T00:00:00Z",
  "affected_channels": ["A", "B", "C"],
  "severity": 0.75,
  "metadata": {
    "device_id": "wave001",
    "location": "main_panel"
  }
}

Response:

{
  "status": "success",
  "event_id": "evt_123",
  "message": "Event ingested successfully"
}

Behavior:

Writes event to events/{device_id}/events.json.log
Broadcasts to all SSE subscribers

`GET /api/v1/events/stream`

Server-Sent Events (SSE) stream for real-time event notifications.

Query Parameters:

device_id (optional): Filter events by device

Response:

Content-Type: text/event-stream
Cache-Control: no-cache

Event Format:

data: {"event_id":"evt_123","event_type":"voltage_sag",...}

: keepalive

Keepalive: Sent every 15 seconds

JavaScript Example:

const eventSource = new EventSource('/api/v1/events/stream?device_id=wave001');

eventSource.onmessage = (event) => {
  const pqEvent = JSON.parse(event.data);
  console.log('Power quality event:', pqEvent);
};

eventSource.onerror = (error) => {
  console.error('SSE error:', error);
};

`GET /api/v1/events`

List all events.

Query Parameters:

device_id (optional): Filter by device
limit (optional): Maximum events (default: 100)

Response:

{
  "events": [...],
  "count": 42
}

`GET /api/v1/events/{id}`

Get event details.

Response:

{
  "event_id": "evt_123",
  "event_type": "voltage_sag",
  "timestamp_start": 1735689600000000,
  "timestamp_end": 1735689650000000,
  "affected_channels": ["A", "B", "C"],
  "severity": 0.75,
  "metadata": {...}
}

`POST /api/v1/events/{id}/assign`

Mark event as assigned for investigation.

Query Parameters:

device_id (required)

Response:

{
  "status": "success",
  "message": "Event evt_123 assigned for device wave001"
}

`POST /api/v1/events/{id}/save`

Save event for later review.

Query Parameters:

device_id (required)

`GET /api/v1/events/assigned`

Get assigned events for a device.

Query Parameters:

device_id (required)

`GET /api/v1/events/saved`

Get saved events for a device.

Query Parameters:

device_id (required)

`GET /api/v1/events/today`

Get events from today.

Query Parameters:

device_id (optional)
limit (optional, default: 100)

`GET /api/v1/events/last7days`

Get events from the last 7 days.

Query Parameters:

device_id (optional)
limit (optional, default: 100)

`GET /api/v1/events/last30days`

Get events from the last 30 days.

Query Parameters:

device_id (optional)
limit (optional, default: 100)

SQL Query Interface

`POST /api/v1/query/sql`

Execute a raw SQL query against the database.

Request Body:

{
  "query": "SELECT * FROM pmon_data WHERE time_us > 1735689600000000 LIMIT 10",
  "device_id": "wave001",
  "limit": 30
}

Security:

Only SELECT statements allowed (INSERT/UPDATE/DELETE rejected)
Default row limit: 30 (prevents accidental large queries)

Response:

{
  "result": "formatted_table_output",
  "rows_returned": 10,
  "truncated": false
}

Example:

curl -X POST http://localhost:8080/api/v1/query/sql \
  -H "Content-Type: application/json" \
  -d '{
    "query": "SELECT time_us, AVRMS, FREQ FROM pmon_data LIMIT 5",
    "device_id": "wave001"
  }'

WebSocket Endpoints

`WS /api/ws/cpow_stream`

Live CPOW waveform streaming via WebSocket.

Connection:

const ws = new WebSocket('ws://localhost:8080/api/ws/cpow_stream');

Protocol:

Connection: Upgrade from HTTP GET request
Data Format: Arrow IPC binary messages
Heartbeat: Server sends ping every 5 seconds
Client Timeout: 10 seconds without pong = disconnect

Message Types:

Binary: Arrow IPC RecordBatch with waveform data
Text: Client commands (future: pause/resume/seek)
Ping/Pong: Keepalive mechanism

Data Flow:

Client connects via WebSocket
Server spawns SharedMemoryReader thread
Reader continuously reads from ArrowSharedMemory (raw_waveform)
Batches are aggregated before sending (see Batch Aggregation below)
Aggregated RecordBatch serialized to Arrow IPC binary format
Binary messages sent to client over WebSocket
Client deserializes Arrow IPC to access waveform data

Batch Aggregation:

The WebSocket stream implements configurable batch aggregation to optimize network efficiency and reduce client CPU overhead:

Sensor packets: 64 rows (2ms @ 32kHz) - written to ArrowSharedMemory
Aggregation factor: 8× (configurable via BATCH_AGGREGATION_FACTOR constant)
WebSocket batches: 512 rows (16ms @ 32kHz) - sent over wire
Message rate: ~62.5 messages/second (vs. ~500/sec without aggregation)
Latency: 16ms (~1 grid cycle @ 60Hz) - excellent for live monitoring

Benefits of 8× aggregation:

Network efficiency: Reduced from ~1,000 Ethernet frames/sec to ~125 frames/sec
CPU overhead: 8× fewer serialization/deserialization cycles
Actor mailbox: Larger buffer headroom (2+ seconds vs. 32ms)
Client performance: Fewer WebSocket messages to process

The aggregation factor can be tuned in cpow_stream.rs:

1 = No aggregation (500 msg/sec, 2ms latency)
8 = Default (62.5 msg/sec, 16ms latency)
125 = Maximum (4 msg/sec, 250ms latency)

Gap Detection:

When the WebSocket reader falls behind and data must be skipped:

A JSON text message is sent: {"type": "gap", "skipped_samples": N}
The frontend can display a visual gap indicator
The reader then catches up to live data

JavaScript Example:

import { tableFromIPC } from 'apache-arrow';

const ws = new WebSocket('ws://localhost:8080/api/ws/cpow_stream');
ws.binaryType = 'arraybuffer';

ws.onopen = () => {
  console.log('WebSocket connected');
};

ws.onmessage = (event) => {
  // Deserialize Arrow IPC binary message
  const table = tableFromIPC(new Uint8Array(event.data));
  console.log(`Received ${table.numRows} waveform samples`);

  // Access data columns
  const timestamps = table.getChild('time_us');
  const voltageA = table.getChild('VA');

  // Process real-time waveforms...
};

ws.onerror = (error) => {
  console.error('WebSocket error:', error);
};

ws.onclose = () => {
  console.log('WebSocket disconnected');
};

Python Example:

import websocket
import pyarrow as pa

def on_message(ws, message):
    # Deserialize Arrow IPC binary
    reader = pa.ipc.open_stream(message)
    table = reader.read_all()
    print(f"Received {len(table)} samples")

def on_error(ws, error):
    print(f"Error: {error}")

def on_close(ws):
    print("WebSocket closed")

def on_open(ws):
    print("WebSocket connected")

ws = websocket.WebSocketApp(
    "ws://localhost:8080/api/ws/cpow_stream",
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)
ws.on_open = on_open
ws.run_forever()

Performance:

Streaming rate: ~100 batches/second
Batch size: Variable (typically 640-32000 samples)
Latency: <50ms from acquisition to client
Logging: Batch count every 100 batches

Data Schemas

PMON Data Schema

time_us: Timestamp(Microsecond, UTC)
device_id: String
FREQ: Float32

# Per-phase voltage (RMS)
AVRMS: Float32
BVRMS: Float32
CVRMS: Float32

# Per-phase current (RMS)
AIRMS: Float32
BIRMS: Float32
CIRMS: Float32
NIRMS: Float32  # Neutral current

# Per-phase real power
AWATT: Float32
BWATT: Float32
CWATT: Float32

# Fundamental voltage (RMS)
AFVRMS: Float32
BFVRMS: Float32
CFVRMS: Float32

# Fundamental current (RMS)
AFIRMS: Float32
BFIRMS: Float32
CFIRMS: Float32

# Fundamental power
AFWATT: Float32
BFWATT: Float32
CFWATT: Float32

# Reactive power
AFVAR: Float32
BFVAR: Float32
CFVAR: Float32

CPOW Data Schema

time_us: Timestamp(Microsecond, UTC)
device_id: String

# Waveform samples (32 kHz sampling)
# Schema varies based on CpowSchemaVariant
# Typically Float32 arrays for voltage and current channels

Error Handling

HTTP Status Codes

200 OK: Success (Arrow IPC body) or no data found (JSON)
400 Bad Request: Invalid query parameters or malformed request
404 Not Found: Device or resource not found
408 Request Timeout: Query exceeded 30-second timeout
500 Internal Server Error: Server-side error

Error Response Format

No Data Found (200 OK):

{
  "status": "success",
  "error": "no_data_found",
  "message": "No data found for device wave001",
  "pipeline_status": "completed",
  "data": [],
  "rows": 0,
  "columns": 0
}

Bad Request (400):

{
  "error": "bad_request",
  "message": "Invalid query parameters: start_time must be ISO 8601 format",
  "pipeline_status": "rejected"
}

Timeout (408):

{
  "error": "timeout",
  "message": "Query timeout after 30 seconds",
  "pipeline_status": "timeout"
}

Internal Error (500):

{
  "error": "internal_error",
  "message": "Database connection failed",
  "pipeline_status": "error"
}

Architecture Details

Request Lifecycle (HTTP REST)

1. HTTP Request
   ↓
2. Actix-web Handler (routes.rs)
   ↓
3. Create DataRequest
   ↓
4. Send to Worker Pool (mpsc channel)
   ↓
5. Worker processes request:
   a. Check LRU cache (60s TTL)
   b. Cache miss → determine data strategy
      - Parquet: Historical data (>24h old)
      - InMemory: Live data (<24h old)
      - Hybrid: Both
   c. Execute DuckDB query via ArrowDuckDBPool
   d. Return RecordBatch
   ↓
6. Worker converts RecordBatch → Arrow IPC binary
   ↓
7. Response sent via oneshot channel
   ↓
8. Handler returns HTTP response
   - Content-Type: application/vnd.apache.arrow.stream
   - Body: Arrow IPC binary

WebSocket Streaming (Live CPOW)

1. Client WebSocket handshake
   ↓
2. CpowStreamSession actor starts
   ↓
3. Spawn SharedMemoryReader thread:
   a. Open ArrowSharedMemory ("raw_waveform")
   b. Loop: read new RecordBatches (64 rows each, 2ms @ 32kHz)
   c. Aggregate batches in WebSocketSink (default 8× = 512 rows)
   d. Send aggregated batches to Actor (62.5/sec vs. 500/sec)
   ↓
4. Actor serializes batches → Arrow IPC
   ↓
5. Send binary messages over WebSocket
   ↓
6. Client receives and deserializes Arrow data

Note: If reader falls behind, gap marker sent as JSON text message

Event Streaming (SSE)

1. Client connects to /api/v1/events/stream
   ↓
2. Server subscribes to broadcast channel
   ↓
3. Events ingested via POST /api/v1/events/ingest:
   a. Write to events/{device_id}/events.json.log
   b. Broadcast to all SSE subscribers
   ↓
4. SSE stream filters by device_id (if specified)
   ↓
5. Events sent as "data: {json}\n\n"
   ↓
6. Keepalive every 15 seconds

Security Considerations

Current Status

Authentication: None implemented

No API keys
No JWT tokens
No OAuth2
Intended for internal/trusted network deployment

CORS: Wide open

Allow any origin
Allow any method
Allow credentials
Suitable for development, should be restricted in production

SQL Injection Protection: Implemented

Only SELECT statements allowed
INSERT/UPDATE/DELETE rejected
Default row limit (30) to prevent resource exhaustion

Path Traversal Protection: Implemented

File names validated (reject .., /, \)
Prevents loading files outside data directories

Production Recommendations

For production deployment:

Add Authentication Middleware:

#![allow(unused)]
fn main() {
.wrap(ApiKeyAuth::new())
.wrap(JwtAuth::new())
}

Restrict CORS:

#![allow(unused)]
fn main() {
Cors::default()
    .allowed_origin("https://app.eq.systems")
    .allowed_methods(vec!["GET", "POST"])
}

Add Rate Limiting:

#![allow(unused)]
fn main() {
.wrap(RateLimiter::new(100, Duration::from_secs(60)))
}

Enable HTTPS/TLS:
- Use reverse proxy (Nginx) with SSL certificates
- Or configure Actix-web with rustls

Performance Tuning

Worker Pool Configuration

Default Settings:

#![allow(unused)]
fn main() {
WorkerConfig {
    pool_size: 1,                  // Number of data workers
    channel_buffer_size: 1000,     // Request queue size
    worker_timeout: Duration::from_secs(30),
    hot_data_retention: Duration::from_secs(24 * 3600),
    query_cache_size: 1000,        // LRU cache entries
    health_check_interval: Duration::from_secs(300),
}
}

Tuning Recommendations:

High Query Volume:

Increase pool_size to 2-4 workers
Increase channel_buffer_size to 5000

Low Latency Priority:

Reduce worker_timeout to 10 seconds
Increase query_cache_size to 10,000

Memory-Constrained:

Reduce query_cache_size to 100
Reduce hot_data_retention to 1 hour

Database Tuning

DuckDB Configuration:

-- Increase memory limit
SET memory_limit='4GB';

-- Enable parallel query execution
SET threads=4;

-- Optimize for Arrow queries
SET enable_object_cache=true;

Deployment

Running the Server

# Development
cd server/crates/synapse-web/services/unified-backend
cargo run

# Production (release build)
cargo run --release

Server Output:

Initializing persistent data provider...
Setting up table management pool...
Spawning background parquet data ingestion task...
Creating worker pool for query processing...
✅ Worker pool created with 1 workers using persistent data storage
Starting HTTP server on 0.0.0.0:8080 with persistent data backend + live streaming
WebSocket live streaming available at /api/ws/cpow_stream

Dependencies

Required:

DuckDB persistent database (auto-created if missing)
Parquet files in PMON_DATA_DIR and CPOW_DATA_DIR
ArrowSharedMemory (raw_waveform) for WebSocket streaming

Optional:

SynapseConfig database for configuration

Environment Variables

Configuration is managed via the synapse-config crate (database-backed).

To customize:

# Set data directories (if needed)
export PMON_DATA_DIR=/path/to/pmon/data
export CPOW_DATA_DIR=/path/to/cpow/data

Troubleshooting

Common Issues

WebSocket Connection Fails:

Check that shared memory (raw_waveform) exists
Verify CPOW acquisition process is running
Check firewall allows WebSocket connections

Query Timeouts:

Increase worker_timeout in WorkerConfig
Optimize DuckDB queries (add indexes)
Reduce query time range or use limit parameter

High Memory Usage:

Reduce query_cache_size
Reduce hot_data_retention
Check for memory leaks in shared memory readers

No Data Returned:

Verify Parquet files exist in data directories
Check background ingestion task completed successfully
Verify time range matches available data

Logging

Enable Debug Logging:

RUST_LOG=debug cargo run

Log Levels:

info: Request/response logging
debug: Worker pool operations, cache hits/misses
trace: Arrow IPC serialization details

CPOW Pipeline - Continuous waveform acquisition and processing
Synapse PMON Pipeline - Power monitoring data flow
Library Crates - Shared library documentation

Summary

The unified-backend provides a high-performance, Arrow-based API for power quality monitoring with:

✅ REST endpoints for historical data queries
✅ WebSocket streaming for real-time waveforms (32 kHz)
✅ Server-Sent Events for power quality event notifications
✅ Zero-copy Arrow IPC format for efficient data transfer
✅ Worker pool architecture with LRU caching
✅ DuckDB persistent storage with background ingestion
✅ Comprehensive error handling

Key Features:

Zero-copy data operations (Arrow IPC)
Sub-50ms latency for live streaming
LRU cache with 60-second TTL
Automatic data strategy selection (Parquet/InMemory/Hybrid)
Language-agnostic client support (Python, JavaScript, Rust)

Endpoints: ~50 total (~25 implemented, ~25 placeholders for future features)

Keyboard shortcuts

Energy Quotient™ Developer Documentation