splp-nodejs

TypeScript/Node.js SDK for the SPLP (Service-oriented Lightweight Protocol) Command Center. Enables applications to publish subscription and exchange requests over Kafka with built-in encryption, sanitization, and distributed tracing.

Features

• Two message patterns — Subscription (one-to-one) and Exchange (one-to-many fan-out)
• End-to-end encryption — Asymmetric encryption via Seal API with circuit breaker protection
• Security hardening — XSS, SQL/NoSQL/command injection protection, prototype pollution guards
• Distributed tracing — OpenTelemetry with trace context propagation across Kafka messages
• mTLS support — Mutual TLS for both Kafka and Seal API connections
• Schema validation — Zod-based validation for all message headers and config

Installation

npm install splp-nodejs
# or
bun add splp-nodejs

Quick Start

Publishing a Request

import { SplpClient } from 'splp-nodejs';

const client = new SplpClient({
  kafka: {
    brokers: ['kafka.example.com:9093'],
    clientId: 'my-app',
    consumerGroupId: 'my-consumer-reply-group',
    ssl: {
      caPath: './certs/kafka/ca.pem',
      certPath: './certs/kafka/client.crt',
      keyPath: './certs/kafka/client.key',
    },
  },
  seal: {
    enabled: true,
    apiBaseUrl: 'https://seal.example.com',
    publicKeys: ['-----BEGIN PUBLIC KEY-----\n...\n-----END PUBLIC KEY-----'],
  },
  tracing: { serviceName: 'my-app' },
});

await client.connect();

// Listen for responses
client.onResponse<{ nik: string; no_kk: string; service: string }>(
  'my-reply-topic',
  async (message) => {
    console.log(message.data);
  },
  async (error) => {
    console.error(`[${error.code}] ${error.message} (requestId: ${error.requestId})`);
  },
);

// Publish a subscription request
await client.publishSubscriptionRequest({
  requestId: crypto.randomUUID(),
  institutionId: 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx',
  namespace: 'dukcapil',
  entity: 'penduduk',
  event: 'verifikasi',
  replyTopic: 'my-reply-topic',
  data: { nik: '3174012501850001', no_kk: '3174012501850002' },
});

Building a Service

import { SplpService } from 'splp-nodejs';

const service = new SplpService({
  serviceId: 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa',
  kafka: {
    brokers: ['kafka.example.com:9093'],
    clientId: 'my-service',
    consumerGroupId: 'my-service-group',  // must be unique per process
  },
  seal: { enabled: false },
  tracing: { serviceName: 'my-service' },
});

await service.connect();

service.onRequest<{ nik: string; no_kk: string }>(
  'my-service-inbox',
  async (message, reply) => {
    const { nik, no_kk } = message.data;
    await reply({ data: { nik, no_kk, service: 'BPS' } });
  },
);

Message Patterns

Subscription

One publisher → Command Center → one registered service → response back to publisher.

Publisher → command-center-inbox → Service Inbox → Service
                                                         ↓
Publisher ← reply-topic ← command-center-inbox ←────────┘

Exchange

One publisher → Command Center → all registered services → all responses back to publisher (one per service).

Publisher → command-center-inbox → Service A Inbox → Service A
                                 → Service B Inbox → Service B
                                                         ↓
Publisher ← reply-topic ← command-center-inbox ←────────┘ (each response arrives separately)

Use publishExchangeRequest() instead of publishSubscriptionRequest() for the exchange pattern.

Consumer Group IDs

Each process that consumes from Kafka must use a unique consumerGroupId. If two processes share the same group ID, Kafka treats them as one consumer group and distributes topic partitions between them — messages may arrive at the wrong process.

// service.ts — subscribes to service inbox
kafka: { consumerGroupId: 'my-service-group', ... }

// consumer.ts — subscribes to reply topic
kafka: { consumerGroupId: 'my-consumer-reply-group', ... }

API Reference

SplpClient

| Method | Description | |--------|-------------| | connect() | Connect to Kafka and initialize tracing | | disconnect() | Graceful shutdown | | publishSubscriptionRequest(params) | Publish a subscription request | | publishExchangeRequest(params) | Publish an exchange request | | onResponse<T>(topic, handler, errorHandler?) | Subscribe to a response topic |

SplpService (extends SplpClient)

| Method | Description | |--------|-------------| | onRequest<T>(topic, handler) | Listen for and handle incoming requests |

Request Parameters

interface SubscriptionRequestParams {
  requestId: string;     // ^[a-zA-Z0-9_-]{1,100}$ — use crypto.randomUUID()
  institutionId: string; // UUID
  namespace: string;     // ^[a-zA-Z0-9._-]{1,200}$
  entity: string;        // ^[a-zA-Z0-9._-]{1,200}$
  event: string;         // ^[a-zA-Z0-9._-]{1,200}$
  replyTopic: string;    // Kafka topic name (max 249 chars)
  data: unknown;         // Any JSON-serializable payload
}

ExchangeRequestParams has the same shape.

Handlers

// Response handler (used with onResponse)
type ResponseHandler<T> = (message: InboundMessage<T>) => Promise<void>;

// Error handler (used with onResponse)
type ErrorHandler = (error: {
  code: string;       // 'INVALID_SCHEMA' | 'DECRYPTION_ERROR'
  message: string;
  requestId: string;
}) => Promise<void>;

// Request handler (used with onRequest)
type RequestHandler<T> = (
  message: InboundMessage<T>,
  reply: (params: { data: unknown }) => Promise<void>
) => Promise<void>;

interface InboundMessage<T> {
  header: MessageHeader;
  data: T;        // Decrypted and sanitized payload
  raw: string;    // Original encrypted ciphertext
}

Configuration

Full Configuration Reference

interface SplpClientConfig {
  kafka: {
    brokers: string[];
    clientId: string;
    consumerGroupId?: string;          // default: '{clientId}-group'
    ssl?: {
      caPath?: string;                 // Path to CA certificate
      certPath: string;                // Path to client certificate (required)
      keyPath: string;                 // Path to client private key (required)
      rejectUnauthorized?: boolean;    // default: false
    };
    publishRetryCount?: number;        // default: 3
    publishRetryBackoffMs?: number;    // default: 500
    maxConcurrentMessages?: number;    // default: 10
    messageTimeoutMs?: number;         // default: 30000
  };
  seal: {
    enabled?: boolean;                 // default: true; false = base64 passthrough
    apiBaseUrl?: string;
    publicKeys?: string[];             // PEM public keys for encryption
    apiTimeout?: number;               // default: 30000
    requestIdPrefix?: string;          // default: 'splp'
    enableLogging?: boolean;           // default: false
    tls?: {
      certPath?: string;               // Path to client certificate
      keyPath?: string;                // Path to client private key
      caPath?: string;                 // Path to CA certificate (optional)
      rejectUnauthorized?: boolean;    // default: false
    };
  };
  tracing?: {
    enabled?: boolean;                 // default: true
    endpoint?: string;                 // OTLP gRPC endpoint
    serviceName: string;               // Required
    sampler?: string;                  // default: 'parentbased_traceid_ratio'
    samplerArg?: number;               // default: 1.0
  };
  sanitizer?: {
    maxStringLength?: number;          // default: 10000
    maxObjectDepth?: number;           // default: 10
    maxArrayLength?: number;           // default: 1000
    strictMode?: boolean;              // default: false
  };
}

interface SplpServiceConfig extends SplpClientConfig {
  serviceId: string;  // UUID — required for SplpService
}

Environment Variables

Copy .env.example to .env and configure:

# --- Identifiers ---
SERVICE_ID=aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa   # UUID for SplpService instances

# --- Kafka ---
KAFKA_BROKERS=localhost:29092                       # Comma-separated broker addresses
KAFKA_CLIENT_ID=my-app
TOPIC_TO_CONSUME=perlinsos.registration.request    # Topic this service subscribes to
SERVICE_CONSUMER_GROUP_ID=consumer-group-registration-dukcapil
KAFKA_PUBLISH_RETRY_COUNT=3
KAFKA_PUBLISH_RETRY_BACKOFF_MS=500
KAFKA_MAX_CONCURRENT_MESSAGES=25
KAFKA_MESSAGE_TIMEOUT_MS=30000

# Kafka SSL/mTLS (omit or set KAFKA_SSL_ENABLED=false for plaintext)
KAFKA_SSL_ENABLED=false
KAFKA_SSL_CA=./certs/kafka/ca.pem
KAFKA_SSL_CERT=./certs/kafka/client.crt
KAFKA_SSL_KEY=./certs/kafka/client.key
KAFKA_SSL_REJECT_UNAUTHORIZED=true

# --- Seal API ---
SEAL_API_URL=https://localhost:2798
SEAL_ENABLED=true
SEAL_PUBLIC_KEY="-----BEGIN PUBLIC KEY-----\nREPLACE_ME\n-----END PUBLIC KEY-----"
SEAL_API_TIMEOUT=30000
SEAL_REQUEST_ID_PREFIX=splp
SEAL_ENABLE_LOGGING=false

# Seal mTLS (omit if not using mTLS)
SEAL_TLS_CERT=./certs/seal/client.crt
SEAL_TLS_KEY=./certs/seal/client.key
SEAL_TLS_REJECT_UNAUTHORIZED=false

# --- Tracing (OpenTelemetry) ---
TRACING_ENABLED=true
OTEL_ENDPOINT=http://localhost:4317
OTEL_SERVICE_NAME=my-app

Certificates

Kafka SSL/mTLS

certs/kafka/
├── ca.pem          # CA certificate
├── client.crt      # Client certificate
└── client.key      # Client private key

Seal API mTLS

certs/seal/
├── client.crt      # Client certificate
├── client.key      # Client private key
└── ca.pem          # CA certificate (optional)

SEAL API Integration

The SEAL (Secure Encryption API Layer) provides asymmetric broadcast encryption for message payloads.

API Endpoints

| Endpoint | Method | Purpose | |----------|--------|---------| | /broadcast/asymmetric/seal/base64 | POST | Encrypt data | | /asymmetric/unseal/base64 | POST | Decrypt single-key ciphertext | | /broadcast/asymmetric/unseal/base64 | POST | Decrypt broadcast ciphertext |

Encryption Flow

JSON Data → Base64 Encode → SEAL API → Ciphertext (stored in message body)

Decryption Flow

The library auto-detects the ciphertext format:

• Single-key: No : separator → tries /asymmetric/unseal/base64, falls back to broadcast endpoint
• Broadcast: Contains : separator → uses /broadcast/asymmetric/unseal/base64

Disabled Mode

When seal.enabled = false, data is base64-encoded as a passthrough (no API call). Useful for local development.

Debug Logging

SEAL_ENABLE_LOGGING=true

Or in config:

seal: { enableLogging: true }

Security

Input Sanitization

All inbound data is automatically sanitized before being passed to handlers:

• XSS — HTML entity encoding, script/iframe/event-handler removal
• SQL injection — Blocks SELECT, INSERT, DROP, UNION, etc.
• NoSQL injection — Blocks $where, $ne, $gt, and similar operators
• Command injection — Blocks shell metacharacters and path traversal patterns
• Prototype pollution — Blocks __proto__, constructor, prototype keys
• Size limits — Strings ≤10,000 chars, objects ≤10 levels deep, arrays ≤1,000 items

Sanitization utilities are also exported for direct use:

import { sanitizeObject, sanitizeXSS, sanitizeInjection, parseJSONSafely } from 'splp-nodejs';

Validation

| Field | Rule | |-------|------| | requestId | ^[a-zA-Z0-9_-]{1,100}$ | | institutionId, serviceId | UUID format | | namespace, entity, event | ^[a-zA-Z0-9._-]{1,200}$ | | Topic names | ^[a-zA-Z0-9._-]{1,249}$, must not start with . or _ |

Error Handling

client.onResponse(
  'my-topic',
  async (message) => { /* ... */ },
  async (error) => {
    // error: { code: string; message: string; requestId: string }
    console.error(`[${error.code}] ${error.message}`);
    // code values: 'INVALID_SCHEMA' | 'DECRYPTION_ERROR'
  },
);

For publish/connect errors thrown synchronously:

import {
  SplpValidationError,
  SplpEncryptionError,
  SplpPublishError,
  SplpConnectionError,
} from 'splp-nodejs';

try {
  await client.publishSubscriptionRequest(params);
} catch (err) {
  if (err instanceof SplpValidationError) {
    console.error('Invalid params:', err.message);
  } else if (err instanceof SplpPublishError) {
    console.error(`Publish failed after ${err.retryCount} retries`);
  }
}

| Error Class | Code | Thrown when | |-------------|------|-------------| | SplpValidationError | VALIDATION_ERROR | Invalid config or request params | | SplpEncryptionError | ENCRYPTION_ERROR | Seal API failure | | SplpPublishError | PUBLISH_ERROR | Kafka publish failure (has retryCount) | | SplpConnectionError | CONNECTION_ERROR | Kafka connect failure | | SplpSecurityError | varies | Security policy violation |

Observability

All operations are traced with OpenTelemetry. Spans include:

| Attribute | Description | |-----------|-------------| | splp.request_id | Request identifier | | splp.type | subscription or exchange | | splp.action | request or response | | splp.namespace | Namespace | | splp.entity | Entity | | splp.event | Event |

Trace context is propagated through Kafka message headers, enabling end-to-end traces across publisher → command center → service → publisher.

Project Structure

splp-nodejs/
├── src/
│   ├── index.ts                    # Public API exports
│   ├── client.ts                   # SplpClient implementation
│   ├── service.ts                  # SplpService implementation
│   ├── config/config.ts            # Defaults and config resolution
│   ├── errors/errors.ts            # Custom error classes
│   ├── kafka/
│   │   ├── kafka-client.ts         # KafkaJS wrapper
│   │   └── trace-propagation.ts    # OTel header injection/extraction
│   ├── message/
│   │   ├── types.ts                # TypeScript interfaces
│   │   ├── envelope.ts             # Message envelope builders
│   │   └── schema.ts               # Zod validation schemas
│   ├── seal/
│   │   ├── seal-encryptor.ts       # Seal API client
│   │   └── circuit-breaker.ts      # Circuit breaker
│   ├── security/
│   │   ├── sanitizer.ts            # XSS / injection sanitization
│   │   ├── safe-json.ts            # Safe JSON parsing
│   │   └── validator.ts            # Input validators
│   └── tracing/
│       ├── tracer.ts               # OTel SDK initialization
│       └── spans.ts                # Span helpers
├── examples/
│   ├── shared-config.ts            # Shared infrastructure config (Kafka, Seal, tracing)
│   ├── subscription/
│   │   ├── publish-request.ts      # Publishes a subscription request
│   │   ├── consumer.ts             # Listens for responses on reply topic
│   │   └── service.ts              # Handles requests and replies
│   └── exchange/
│       ├── publish-request.ts      # Publishes an exchange request
│       ├── consumer.ts             # Listens for responses on reply topic
│       ├── service-a.ts            # Service A handler
│       └── service-b.ts            # Service B handler
├── tests/
├── certs/
│   ├── kafka/
│   └── seal/
└── .env.example

Running the Examples

# Terminal 1 — service that handles requests
bun run examples/subscription/service.ts

# Terminal 2 — consumer that receives responses
bun run examples/subscription/consumer.ts

# Terminal 3 — publish a request
bun run examples/subscription/publish-request.ts

For the exchange flow (fan-out to multiple services):

# Terminal 1
bun run examples/exchange/service-a.ts

# Terminal 2
bun run examples/exchange/service-b.ts

# Terminal 3 — consumer receives one response per service
bun run examples/exchange/consumer.ts

# Terminal 4
bun run examples/exchange/publish-request.ts

Development

# Install dependencies
bun install

# Type check
bun run typecheck

# Build
bun run build

# Run tests
bun test