@hamicek/noex-server

WebSocket server for @hamicek/noex-store and @hamicek/noex-rules built on @hamicek/noex GenServer supervision.

Features

• GenServer per connection with simple_one_for_one supervision and automatic cleanup
• Full proxy for noex-store: CRUD, reactive query subscriptions, multi-bucket transactions
• Optional noex-rules proxy: emit events, manage facts, subscribe to rule matches
• Token-based authentication with pluggable validation and per-operation permissions
• Rate limiting, heartbeat ping/pong, and write-buffer backpressure
• Graceful shutdown with client notification and configurable grace period
• Connection registry with real-time stats and per-connection metadata
• JSON-over-WebSocket protocol (version 1.0.0) with request/response correlation and push channels

Installation

npm install @hamicek/noex-server

Requires @hamicek/noex and @hamicek/noex-store as peer dependencies and Node.js >= 20. @hamicek/noex-rules is an optional peer dependency.

Quick Start

import { Store } from '@hamicek/noex-store';
import { NoexServer } from '@hamicek/noex-server';

const store = await Store.start({ name: 'my-store' });

store.defineBucket('users', {
  key: 'id',
  schema: {
    id:   { type: 'string', generated: 'uuid' },
    name: { type: 'string', required: true },
    role: { type: 'string', default: 'user' },
  },
});

store.defineQuery('all-users', async (ctx) => ctx.bucket('users').all());

const server = await NoexServer.start({
  port: 8080,
  store,
});

console.log(`Listening on ws://localhost:${server.port}`);

A client connects over WebSocket and sends JSON messages:

// Insert a record
→ { "id": 1, "type": "store.insert", "bucket": "users", "data": { "name": "Alice" } }
← { "id": 1, "type": "result", "data": { "id": "a1b2c3", "name": "Alice", "role": "user", "_version": 1, ... } }

// Subscribe to a reactive query
→ { "id": 2, "type": "store.subscribe", "query": "all-users" }
← { "id": 2, "type": "result", "data": { "subscriptionId": "sub-1" } }
← { "type": "push", "channel": "subscription", "subscriptionId": "sub-1", "data": [...] }

API

NoexServer

NoexServer.start(config): Promise<NoexServer>

Creates and starts the server. Initializes the HTTP server, WebSocket upgrade handler, connection supervisor, and optional rate limiter.

const server = await NoexServer.start({
  port: 8080,
  store,
  rules: engine,        // optional
  auth: { ... },        // optional
  rateLimit: { ... },   // optional
});

server.stop(options?): Promise<void>

Gracefully stops the server.

Option	Type	Default	Description
gracePeriodMs	number	0	Time to wait for clients to disconnect after sending a shutdown notification

await server.stop({ gracePeriodMs: 5000 });

server.port: number

The port the server is listening on. Useful when starting with port: 0 for tests.

server.connectionCount: number

Number of active WebSocket connections.

server.isRunning: boolean

Whether the server is currently accepting connections.

server.getConnections(): ConnectionInfo[]

Returns metadata for all active connections: remote address, auth status, subscription counts, connected timestamp.

server.getStats(): Promise<ServerStats>

Aggregated statistics including connection counts, uptime, feature flags, and underlying store/rules stats.

const stats = await server.getStats();
// {
//   name: 'noex-server',
//   port: 8080,
//   connectionCount: 42,
//   uptimeMs: 360000,
//   authEnabled: true,
//   rateLimitEnabled: true,
//   rulesEnabled: false,
//   connections: { active: 42, authenticated: 40, totalStoreSubscriptions: 120, ... },
//   store: { ... },
//   rules: null,
// }

---

Configuration

interface ServerConfig {
  store: Store;                        // required — noex-store instance
  rules?: RuleEngine;                  // optional — noex-rules instance
  port?: number;                       // default: 8080
  host?: string;                       // default: '0.0.0.0'
  path?: string;                       // default: '/' (WebSocket endpoint)
  maxPayloadBytes?: number;            // default: 1 MB
  auth?: AuthConfig;                   // when omitted, auth is disabled
  rateLimit?: RateLimitConfig;         // when omitted, rate limiting is disabled
  heartbeat?: HeartbeatConfig;         // default: 30 s interval, 10 s timeout
  backpressure?: BackpressureConfig;   // default: 1 MB limit, 0.8 high water mark
  name?: string;                       // default: 'noex-server'
}

Authentication

const server = await NoexServer.start({
  store,
  auth: {
    validate: async (token) => {
      const payload = verifyJwt(token);
      if (!payload) return null;
      return {
        userId: payload.sub,
        roles: payload.roles ?? ['user'],
        expiresAt: payload.exp * 1000,
      };
    },
    required: true, // default when auth is configured
    permissions: {
      check: (session, operation, resource) => {
        if (session.roles.includes('admin')) return true;
        if (operation === 'store.clear') return false;
        return true;
      },
    },
  },
});

When auth.required is true, clients must send auth.login before any other operation. Session expiration is checked on every request.

Rate Limiting

rateLimit: {
  maxRequests: 200,   // requests per window
  windowMs: 60_000,   // sliding window duration
}

Uses @hamicek/noex RateLimiter GenServer. Key is session.userId for authenticated connections, remote IP address otherwise.

Heartbeat

heartbeat: {
  intervalMs: 30_000,  // how often to send ping
  timeoutMs: 10_000,   // how long to wait for pong
}

The server sends ping messages at the configured interval. If no pong is received within timeoutMs, the connection is closed with code 4001.

Backpressure

backpressure: {
  maxBufferedBytes: 1_048_576,  // 1 MB
  highWaterMark: 0.8,           // pause push at 80%
}

When the WebSocket write buffer exceeds the high water mark, push messages are paused to prevent memory exhaustion on slow clients.

---

Protocol

All messages are JSON objects sent as WebSocket text frames. Protocol version: 1.0.0.

Connection Lifecycle

1. Client connects via WebSocket
2. Server sends a welcome message with protocol version and auth requirements
3. If auth is required, client sends auth.login with a token
4. Client sends requests, server responds with results or errors
5. For subscriptions, server sends asynchronous push messages
6. Server sends periodic ping, client responds with pong
7. Either side can close the connection; server cleans up all subscriptions

Message Types

Request (client -> server):

{ "id": 1, "type": "store.insert", "bucket": "users", "data": { "name": "Alice" } }

Response (server -> client):

{ "id": 1, "type": "result", "data": { ... } }
{ "id": 1, "type": "error", "code": "VALIDATION_ERROR", "message": "...", "details": { ... } }

Push (server -> client, no request correlation):

{ "type": "push", "channel": "subscription", "subscriptionId": "sub-1", "data": [...] }
{ "type": "push", "channel": "event", "subscriptionId": "sub-2", "data": { "topic": "...", "event": { ... } } }

System (server -> client):

{ "type": "welcome", "version": "1.0.0", "serverTime": 1706745600000, "requiresAuth": true }
{ "type": "ping", "timestamp": 1706745600000 }
{ "type": "system", "event": "shutdown", "gracePeriodMs": 5000 }

Operations

Store

Operation	Description	Payload fields
store.insert	Insert a record	bucket, data
store.get	Get by primary key	bucket, key
store.update	Update a record	bucket, key, data
store.delete	Delete a record	bucket, key
store.all	All records	bucket
store.where	Filter records	bucket, filter
store.findOne	First match	bucket, filter
store.count	Count records	bucket, filter?
store.first	First N records	bucket, n
store.last	Last N records	bucket, n
store.paginate	Cursor pagination	bucket, limit, after?
store.clear	Clear all records	bucket
store.sum	Sum a numeric field	bucket, field, filter?
store.avg	Average a numeric field	bucket, field, filter?
store.min	Minimum value	bucket, field, filter?
store.max	Maximum value	bucket, field, filter?
store.subscribe	Subscribe to reactive query	query, params?
store.unsubscribe	Cancel subscription	subscriptionId
store.transaction	Atomic multi-bucket transaction	operations
store.buckets	List defined buckets	—
store.stats	Store statistics	—

Transactions

Send multiple operations atomically:

{
  "id": 10,
  "type": "store.transaction",
  "operations": [
    { "op": "get", "bucket": "users", "key": "user-1" },
    { "op": "update", "bucket": "users", "key": "user-1", "data": { "credits": 400 } },
    { "op": "insert", "bucket": "logs", "data": { "action": "credit_update" } }
  ]
}

Supported ops: get, insert, update, delete, where, findOne, count.

Rules

Available only when rules is configured. Returns RULES_NOT_AVAILABLE otherwise.

Operation	Description	Payload fields
rules.emit	Emit an event	topic, data, correlationId?
rules.setFact	Set a fact	key, value
rules.getFact	Get a fact	key
rules.deleteFact	Delete a fact	key
rules.queryFacts	Query facts by pattern	pattern
rules.getAllFacts	Get all facts	—
rules.subscribe	Subscribe to rule events	pattern
rules.unsubscribe	Cancel subscription	subscriptionId
rules.stats	Engine statistics	—

Auth

Operation	Description	Payload fields
auth.login	Authenticate with token	token
auth.logout	End session	—
auth.whoami	Current session info	—

Error Codes

Code	Description
PARSE_ERROR	Invalid JSON
INVALID_REQUEST	Missing id or type
UNKNOWN_OPERATION	Unsupported operation type
VALIDATION_ERROR	Schema validation failed
NOT_FOUND	Record or subscription not found
ALREADY_EXISTS	Duplicate key or unique constraint violation
CONFLICT	Transaction version conflict
UNAUTHORIZED	Authentication required or token invalid
FORBIDDEN	Insufficient permissions
RATE_LIMITED	Rate limit exceeded
BACKPRESSURE	Write buffer full, slow down
INTERNAL_ERROR	Unexpected server error
BUCKET_NOT_DEFINED	Bucket does not exist
QUERY_NOT_DEFINED	Reactive query not defined
RULES_NOT_AVAILABLE	Rule engine not configured

---

Architecture

NoexServer
└── ConnectionSupervisor (simple_one_for_one)
    ├── ConnectionServer #1  (GenServer per WebSocket)
    ├── ConnectionServer #2
    └── ConnectionServer #N

Each WebSocket connection is managed by a dedicated ConnectionServer GenServer. The supervisor uses the temporary restart strategy — crashed connections are cleaned up (all subscriptions unsubscribed, WebSocket closed) but not restarted.

The request pipeline for each message:

1. JSON parse and validate
2. Authentication check (if configured)
3. Rate limit check (if configured)
4. Route to store proxy or rules proxy
5. Serialize result or map error

---

License

MIT