@progalaxyelabs/stonescriptphp-sse

Generic SSE (Server-Sent Events) server with JWKS authentication and plugin hooks. Mounts onto any Express application.

Features

• JWKS authentication — strict RS256/ES256 allowlist, audience+issuer pinning, 60s clock skew
• Channel subscriptions — clients subscribe to named channels
• User-targeted events — publishToUser(userId, event) reaches all of a user's connections
• Reconnect support — ring-buffer per channel; Last-Event-ID triggers replay
• Plugin hooks — override auth, authorization, connect/disconnect logic
• Heartbeat — configurable comment-based keepalive

Installation

npm install @progalaxyelabs/stonescriptphp-sse

Express 5 is a peer dependency:

npm install express@^5

Quick start

import express from 'express';
import { createSseServer } from '@progalaxyelabs/stonescriptphp-sse';

const app = express();

const sseServer = createSseServer({
  jwks: {
    url:      process.env.JWKS_URL,       // e.g. https://auth.example.com/.well-known/jwks.json
    issuer:   process.env.JWT_ISSUER,     // e.g. https://auth.example.com
    audience: process.env.JWT_AUDIENCE,   // e.g. my-app
  },
  heartbeatInterval: 30_000,
});

app.use('/sse', sseServer.router);

app.listen(3000, () => console.log('Listening on :3000'));

// Publish from anywhere in your business logic:
sseServer.publish('low-stock', { type: 'low_stock', data: { item_id: 42, qty: 2 } });
sseServer.publishToUser('user-123', { type: 'alert', data: { message: 'Low stock warning' } });

Client (browser)

// Browsers send the Authorization header via EventSource only in fetch-based polyfills.
// The recommended approach is a token query parameter:
const token = getAccessToken(); // your auth flow
const es = new EventSource(`/sse?token=${token}&channel=low-stock`);

es.addEventListener('low_stock', (e) => {
  console.log('Low stock alert:', JSON.parse(e.data));
});

es.addEventListener('connected', (e) => {
  console.log('SSE connected', JSON.parse(e.data));
});

Plugin hooks

Override any hook by passing a hooks object to createSseServer:

const sseServer = createSseServer({
  jwks: { /* ... */ },
  hooks: {
    /**
     * Custom authentication — e.g. add extra user fields from your DB.
     * Receives the raw token string. Must return a user payload or throw with status=401.
     */
    authenticateSubscriber: async (token) => {
      const payload = await myJwksVerifier.verify(token); // or use the default
      const user = await db.users.findById(payload.sub);
      return { ...payload, role: user.role };
    },

    /**
     * Channel authorization — return false to block the subscription.
     * Default: allow all channels.
     */
    authorizeChannel: async (user, channel) => {
      if (channel.startsWith('admin:')) return user.role === 'admin';
      if (channel.startsWith('user:')) return channel === `user:${user.sub}`;
      return true;
    },

    onConnect:    async (client) => console.log('connect',    client.id),
    onDisconnect: async (client) => console.log('disconnect', client.id),
  },
});

API reference

createSseServer(options)

Returns { router, publish, publishToUser, stop, clients }.

| Option | Type | Default | Description | |--------|------|---------|-------------| | jwks.url | string | JWKS_URL env | Remote JWKS endpoint | | jwks.issuer | string | JWT_ISSUER env | Expected iss claim | | jwks.audience | string | JWT_AUDIENCE env | Expected aud claim | | heartbeatInterval | number | 30000 (or SSE_HEARTBEAT_INTERVAL env) | ms between heartbeat comments | | hooks | object | defaults | Plugin hooks (see above) |

sseServer.router

Express Router. Mount with app.use('/sse', sseServer.router).

Routes:

• GET / — SSE subscribe endpoint
• GET /health — liveness probe (no auth, returns {"status":"healthy","clients":N})

Query parameters for GET /:

• channel (repeatable) — channels to subscribe to
• token — JWT bearer token (alternative to Authorization: Bearer header)

sseServer.publish(channel, event)

Broadcast an event to all subscribers of channel. Always buffered for reconnect replay.

sseServer.publish('notifications', {
  type: 'new_order',
  data: { order_id: 99, total: 250 }
});

sseServer.publishToUser(userId, event)

Send an event to all connections belonging to a specific user (sub claim).

sseServer.publishToUser('user-42', {
  type: 'alert',
  data: { message: 'Your order shipped!' }
});

sseServer.stop()

Stop the heartbeat timer. Call during graceful shutdown.

sseServer.clients

Map<clientId, { id, userId, channels, res, connectedAt }> — live connected clients.

Environment variables

| Variable | Required | Default | Description | |----------|----------|---------|-------------| | JWKS_URL | ✓ (or pass in jwks.url) | — | JWKS endpoint URL | | JWT_ISSUER | ✓ (or pass in jwks.issuer) | — | JWT issuer | | JWT_AUDIENCE | ✓ (or pass in jwks.audience) | — | JWT audience | | SSE_HEARTBEAT_INTERVAL | — | 30000 | Heartbeat interval (ms) | | PORT | — | app decides | Port (not used by this lib directly) | | CORS_ORIGINS | — | app decides | CORS (configure on your Express app) |

JWT security

• Allowed algorithms: RS256, ES256 only
• Rejected: HS256, none, any unknown algorithm
• Audience pinning: enforced via audience config
• Issuer pinning: enforced via issuer config
• Clock skew: max 60 seconds

Any token that fails these checks returns a 401 before JWKS lookup (where applicable).

Reconnect (Last-Event-ID)

The server maintains a ring buffer of the last 100 events per channel. When a client reconnects and sends Last-Event-ID: N, all buffered events with id > N are replayed immediately after the connected event.

Browser EventSource handles Last-Event-ID automatically.

License

MIT