kyuno/lightweight-sdk
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Ajouter @lightweight-sdk/service-discovery (packages/server) #3

Open
kyuno wants to merge 1 commit from feat/service-discovery-package into feat/service-registry-package
pull from: feat/service-discovery-package
merge into: kyuno:feat/service-registry-package
Conversation 0 Commits 1 Files changed 30 +1592 -5
30 changed files with 1592 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
.changeset/add-service-discovery-package.md Normal file
View file

@ -0,0 +1,5 @@
---
"@lightweight-sdk/service-discovery": minor
---
Ajout du package `@lightweight-sdk/service-discovery` : agrégateur d'API reverse-proxy avec enregistrement et découverte de services au runtime. `DiscoveryClient` (`./client`) enregistre un service et maintient un heartbeat périodique ; `ServiceAggregator` (`./aggregator`) expose des handlers Node natifs à monter dans un serveur existant (Express, Fastify ou `node:http`), route les requêtes entrantes vers le bon service selon le préfixe de nom, et évince les enregistrements expirés (TTL). Zéro dépendance framework HTTP.

114
packages/server/service-discovery/README.md Normal file
View file

@ -0,0 +1,114 @@
# @lightweight-sdk/service-discovery
Agrégateur d'API reverse-proxy avec enregistrement et découverte de services au runtime. Package `server` : deux points d'entrée séparés (`./client` et `./aggregator`), zéro dépendance framework (`node:http`/`node:https`/`fetch` uniquement).
## Installation
```bash
pnpm add @lightweight-sdk/service-discovery
```
## Usage
### Un service qui s'enregistre
```ts
import { DiscoveryClient } from '@lightweight-sdk/service-discovery/client';
const client = new DiscoveryClient({
aggregatorUrl: 'http://aggregator.internal:8080',
serviceName: 'my-service',
baseUrl: 'http://my-service.internal:3000',
endpoints: [{ method: 'GET', path: '/users' }],
sharedSecret: process.env.DISCOVERY_SECRET!,
});
await client.start(); // enregistrement initial + heartbeat périodique
process.on('SIGTERM', () => {
void client.stop(); // désinscription best-effort, arrête le heartbeat
});
```
### L'agrégateur
```ts
import { ServiceAggregator } from '@lightweight-sdk/service-discovery/aggregator';
const aggregator = new ServiceAggregator({ sharedSecret: process.env.DISCOVERY_SECRET! });
aggregator.start(); // démarre le balayage d'expiration (TTL)
```
Le cœur s'expose comme des **handlers Node natifs** (`(req: IncomingMessage, res: ServerResponse) => void | Promise<void>`) — à monter toi-même dans ton serveur existant :
**Express** (`req`/`res` Express *sont* des `IncomingMessage`/`ServerResponse`) :
```ts
app.post('/discovery/register', aggregator.registrationHandler);
app.post('/discovery/deregister', aggregator.deregistrationHandler);
app.all('*', aggregator.proxyHandler); // monté en dernier, après tes propres routes
```
⚠️ `'*'` seul est spécifique à Express 4. Express 5 a cassé ce wildcard non-nommé (`path-to-regexp` v8) — il faut `'/*splat'`.
**Fastify** (`reply.hijack()` indique à Fastify de ne plus toucher `reply` après cet appel) :
```ts
fastify.post('/discovery/register', async (request, reply) => {
reply.hijack();
await aggregator.registrationHandler(request.raw, reply.raw).catch((e) => request.log.error(e));
});
fastify.all('/*', async (request, reply) => {
reply.hijack();
await aggregator.proxyHandler(request.raw, reply.raw).catch((e) => request.log.error(e));
});
```
**`node:http` brut** :
```ts
const server = http.createServer((req, res) => {
if (req.method === 'POST' && req.url === '/discovery/register') return void aggregator.registrationHandler(req, res);
if (req.method === 'POST' && req.url === '/discovery/deregister') return void aggregator.deregistrationHandler(req, res);
void aggregator.proxyHandler(req, res);
});
```
Une requête `GET http://aggregator.internal:8080/my-service/users` est alors relayée vers `http://my-service.internal:3000/users`.
## Concepts
- **Enregistrement = heartbeat** : `DiscoveryClient.start()` envoie un enregistrement initial puis répète le même appel à intervalle régulier (`heartbeatIntervalMs`). C'est aussi le mécanisme de retry : un échec (agrégateur pas encore démarré, panne transitoire) est juste loggé, le prochain tick réessaie — pas de logique de retry séparée.
- **TTL côté agrégateur** : chaque enregistrement expire après `ttlMs` (par défaut 3x l'intervalle de heartbeat par défaut, pour tolérer 1-2 heartbeats manqués) sans renouvellement. Un balayage périodique (`sweepIntervalMs`) évince les entrées expirées. Pas de persistance : si l'agrégateur redémarre, chaque service se réenregistre au heartbeat suivant.
- **Routage par préfixe de nom de service** : `GET /mon-service/users` est matché contre l'endpoint `GET /users` déclaré par le service `mon-service`, puis relayé vers `${baseUrl}/users`.
- **Secret partagé** : toute requête d'enregistrement/désinscription doit présenter `Authorization: Bearer <secret>`. Comparaison en temps constant (`crypto.timingSafeEqual`).
## Options
### `DiscoveryClientOptions`
| Option | Défaut | Description |
|---|---|---|
| `heartbeatIntervalMs` | `15000` | Intervalle entre deux heartbeats |
| `requestTimeoutMs` | `5000` | Timeout de chaque requête d'enregistrement/heartbeat/désinscription |
| `registerPath` | `/discovery/register` | Chemin d'enregistrement sur l'agrégateur |
| `deregisterPath` | `/discovery/deregister` | Chemin de désinscription sur l'agrégateur |
### `ServiceAggregatorOptions`
| Option | Défaut | Description |
|---|---|---|
| `ttlMs` | `45000` | Durée de validité d'un enregistrement sans heartbeat |
| `sweepIntervalMs` | `10000` | Fréquence du balayage d'éviction |
| `proxyTimeoutMs` | `10000` | Timeout d'**inactivité socket** (pas une durée totale) vers le backend |
| `maxRegistrationBodyBytes` | `65536` | Taille max acceptée pour un payload d'enregistrement/désinscription |
## Limites connues
- **Pas de matching `:param`** — le routage fait un match exact du chemin (`/users` ≠ `/users/:id`). Une vraie API REST a presque toujours ce genre de routes ; l'ajout d'un matching segment par segment façon Express est une extension future volontairement différée.
- **Pas de secret par service** — un seul secret partagé pour toute la registry.
- **Pas de persistance** — décision assumée, voir "Concepts" ci-dessus.
- **Collision de nom** — un service enregistré sous un nom identique au segment de chemin où tu montes les routes de contrôle (ex. un service nommé `discovery` si tu montes sous `/discovery/*`) verrait ses propres endpoints inaccessibles, la route de contrôle étant toujours prioritaire. À éviter par convention de nommage.
- **Risque à connaître (pas corrigé)** : dans le threat model visé (homelab, services de confiance), un service compromis pourrait déclarer un `baseUrl` arbitraire et transformer l'agrégateur en proxy ouvert vers ce point (SSRF-like). Aucun enforcement n'est prévu — à garder en tête si le threat model change.
## API
- `@lightweight-sdk/service-discovery` (racine) — protocole partagé uniquement : `RegistrationPayload`, `ServiceEndpoint`, `HttpMethod`, `RegistrationAck`, `DeregistrationPayload`.
- `@lightweight-sdk/service-discovery/client` — `DiscoveryClient`, `DiscoveryClientOptions`.
- `@lightweight-sdk/service-discovery/aggregator` — `ServiceAggregator` (`registrationHandler`, `deregistrationHandler`, `proxyHandler`, `start()`, `stop()`, `getRegisteredServiceNames()`), `ServiceAggregatorOptions`.

4
packages/server/service-discovery/aggregator/package.json Normal file
View file

@ -0,0 +1,4 @@
{
"main": "../dist/aggregator/index.js",
"types": "../dist/aggregator/index.d.ts"
}

4
packages/server/service-discovery/client/package.json Normal file
View file

@ -0,0 +1,4 @@
{
"main": "../dist/client/index.js",
"types": "../dist/client/index.d.ts"
}

32
packages/server/service-discovery/package.json Normal file
View file

@ -0,0 +1,32 @@
{
"name": "@lightweight-sdk/service-discovery",
"version": "0.0.1",
"description": "Agrégateur d'API reverse-proxy avec enregistrement et découverte de services au runtime.",
"main": "dist/index.js",
"types": "dist/index.d.ts",
"files": [
"dist",
"client",
"aggregator"
],
"engines": {
"node": ">=22"
},
"publishConfig": {
"access": "restricted",
"registry": "https://git.dev.kyuno.fr/api/packages/kyuno/npm/"
},
"typedocOptions": {
"entryPoints": [
"src/index.ts",
"src/client/index.ts",
"src/aggregator/index.ts"
],
"readme": "README.md"
},
"devDependencies": {
"@types/node": "catalog:",
"typescript": "catalog:",
"vitest": "catalog:"
}
}

59
packages/server/service-discovery/src/aggregator/http-utils.test.ts Normal file
View file

@ -0,0 +1,59 @@
import type { IncomingMessage } from 'node:http';
import { Readable } from 'node:stream';
import { describe, expect, it } from 'vitest';
import { filterHopByHop, readJsonBody, timingSafeEqualStrings } from './http-utils';
function fakeRequest(chunks: string[]): IncomingMessage {
return Readable.from(
chunks.map((chunk) => Buffer.from(chunk, 'utf8')),
) as unknown as IncomingMessage;
}
describe('timingSafeEqualStrings', () => {
it('returns true for equal strings', () => {
expect(timingSafeEqualStrings('shh', 'shh')).toBe(true);
});
it('returns false for different strings of the same length', () => {
expect(timingSafeEqualStrings('shh1', 'shh2')).toBe(false);
});
it('returns false for strings of different lengths', () => {
expect(timingSafeEqualStrings('short', 'a-much-longer-string')).toBe(false);
});
});
describe('readJsonBody', () => {
it('parses a valid JSON body', async () => {
await expect(readJsonBody(fakeRequest(['{"hello":"world"}']), 1024)).resolves.toEqual({
hello: 'world',
});
});
it('resolves to undefined for an empty body', async () => {
await expect(readJsonBody(fakeRequest([]), 1024)).resolves.toBeUndefined();
});
it('throws on malformed JSON', async () => {
await expect(readJsonBody(fakeRequest(['not json']), 1024)).rejects.toThrow();
});
it('throws when the body exceeds the byte limit', async () => {
await expect(
readJsonBody(fakeRequest(['{"padding":"'.padEnd(50, 'x') + '"}']), 10),
).rejects.toThrow(/exceeds/);
});
});
describe('filterHopByHop', () => {
it('removes hop-by-hop headers while keeping the rest', () => {
const filtered = filterHopByHop({
connection: 'keep-alive',
'transfer-encoding': 'chunked',
'content-type': 'application/json',
'x-custom': 'value',
});
expect(filtered).toEqual({ 'content-type': 'application/json', 'x-custom': 'value' });
});
});

68
packages/server/service-discovery/src/aggregator/http-utils.ts Normal file
View file

@ -0,0 +1,68 @@
import { timingSafeEqual } from 'node:crypto';
import type { IncomingHttpHeaders, IncomingMessage } from 'node:http';
/** Headers that are only meaningful between two adjacent hops (RFC 7230 §6.1) and must never be blindly forwarded by a proxy. */
const HOP_BY_HOP_HEADERS = new Set([
'connection',
'keep-alive',
'proxy-authenticate',
'proxy-authorization',
'te',
'trailers',
'transfer-encoding',
'upgrade',
]);
/**
* Compares two strings in constant time, to avoid leaking information about
* a secret's value through response-time differences. Costs nothing (uses
* only `node:crypto`, already a runtime dependency of this package) so
* there is no reason to prefer a plain `===` here.
* @param a first string to compare
* @param b second string to compare
* @returns `true` if `a` and `b` are exactly equal
*/
export function timingSafeEqualStrings(a: string, b: string): boolean {
const bufA = Buffer.from(a);
const bufB = Buffer.from(b);
return bufA.length === bufB.length && timingSafeEqual(bufA, bufB);
}
/**
* Reads and JSON-parses the body of an incoming request, enforcing a
* maximum size to avoid unbounded memory use from a misbehaving or
* malicious caller.
* @param req the incoming request to read
* @param maxBytes maximum accepted body size, in bytes
* @returns the parsed JSON body
* @throws {Error} if the body exceeds `maxBytes` or is not valid JSON
*/
export async function readJsonBody(req: IncomingMessage, maxBytes: number): Promise<unknown> {
const chunks: Buffer[] = [];
let totalBytes = 0;
for await (const chunk of req as AsyncIterable<Buffer>) {
totalBytes += chunk.length;
if (totalBytes > maxBytes) {
throw new Error(`request body exceeds the ${maxBytes} byte limit`);
}
chunks.push(chunk);
}
const raw = Buffer.concat(chunks).toString('utf8');
return raw.length === 0 ? undefined : JSON.parse(raw);
}
/**
* Returns a copy of `headers` with hop-by-hop headers removed, safe to
* forward across a proxy hop in either direction.
* @param headers the headers to filter
* @returns a new headers object without hop-by-hop entries
*/
export function filterHopByHop(headers: IncomingHttpHeaders): IncomingHttpHeaders {
const filtered: IncomingHttpHeaders = {};
for (const [key, value] of Object.entries(headers)) {
if (!HOP_BY_HOP_HEADERS.has(key.toLowerCase())) {
filtered[key] = value;
}
}
return filtered;
}

3
packages/server/service-discovery/src/aggregator/index.ts Normal file
View file

@ -0,0 +1,3 @@
export type { HttpMethod, RegistrationAck, ServiceEndpoint } from '../types';
export type { ServiceAggregatorOptions } from './service-aggregator';
export { ServiceAggregator } from './service-aggregator';

141
packages/server/service-discovery/src/aggregator/proxy-request.test.ts Normal file
View file

@ -0,0 +1,141 @@
import type { IncomingMessage, ServerResponse } from 'node:http';
import { createServer as createHttpServer, Agent as HttpAgent } from 'node:http';
import { Agent as HttpsAgent } from 'node:https';
import { afterEach, describe, expect, it } from 'vitest';
import type { EphemeralServer } from '../test-support/ephemeral-http-server';
import { startEphemeralServer } from '../test-support/ephemeral-http-server';
import { proxyRequest } from './proxy-request';
import type { RegisteredService } from './registered-service';
import type { RouteMatch } from './route-matching';
const httpAgent = new HttpAgent({ keepAlive: false });
const httpsAgent = new HttpsAgent({ keepAlive: false });
/** Starts a proxy-fronting server that relays every request to `match` via `proxyRequest`. */
async function startProxyFront(match: RouteMatch, timeoutMs = 2000): Promise<EphemeralServer> {
return startEphemeralServer((req: IncomingMessage, res: ServerResponse) => {
proxyRequest(req, res, match, { timeoutMs, httpAgent, httpsAgent });
});
}
function matchFor(service: RegisteredService, remainderPath: string): RouteMatch {
return { serviceName: 'backend', service, remainderPath };
}
describe('proxyRequest', () => {
const servers: EphemeralServer[] = [];
afterEach(async () => {
await Promise.all(servers.splice(0).map((s) => s.close()));
});
it('relays a GET request, stripping the service prefix and forwarding query/headers', async () => {
const backend = await startEphemeralServer((req, res) => {
res.writeHead(200, {
'content-type': 'application/json',
'x-echo': String(req.headers['x-request-id']),
});
res.end(JSON.stringify({ path: req.url }));
});
servers.push(backend);
const service: RegisteredService = {
baseUrl: backend.url,
endpoints: [],
expiresAt: Date.now() + 60_000,
};
const front = await startProxyFront(matchFor(service, '/users'));
servers.push(front);
const response = await fetch(`${front.url}/my-service/users?active=true`, {
headers: { 'x-request-id': 'abc' },
});
expect(response.status).toBe(200);
expect(response.headers.get('x-echo')).toBe('abc');
expect(await response.json()).toEqual({ path: '/users?active=true' });
});
it('relays a POST request with a JSON body intact', async () => {
const backend = await startEphemeralServer((req, res) => {
const chunks: Buffer[] = [];
req.on('data', (chunk: Buffer) => chunks.push(chunk));
req.on('end', () => {
res.writeHead(201, { 'content-type': 'application/json' });
res.end(Buffer.concat(chunks));
});
});
servers.push(backend);
const service: RegisteredService = {
baseUrl: backend.url,
endpoints: [],
expiresAt: Date.now() + 60_000,
};
const front = await startProxyFront(matchFor(service, '/users'));
servers.push(front);
const response = await fetch(`${front.url}/my-service/users`, {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ name: 'ada' }),
});
expect(response.status).toBe(201);
expect(await response.json()).toEqual({ name: 'ada' });
});
it('sends the backend host in the Host header, not the front-facing aggregator host', async () => {
const backend = await startEphemeralServer((req, res) => {
res.writeHead(200, { 'content-type': 'text/plain' });
res.end(req.headers.host ?? '');
});
servers.push(backend);
const service: RegisteredService = {
baseUrl: backend.url,
endpoints: [],
expiresAt: Date.now() + 60_000,
};
const front = await startProxyFront(matchFor(service, '/'));
servers.push(front);
const response = await fetch(`${front.url}/my-service`);
expect(await response.text()).toBe(`127.0.0.1:${backend.port}`);
});
it('responds 502 when the backend refuses the connection', async () => {
const closed = await startEphemeralServer(() => {});
const deadPort = closed.port;
await closed.close();
const service: RegisteredService = {
baseUrl: `http://127.0.0.1:${deadPort}`,
endpoints: [],
expiresAt: Date.now() + 60_000,
};
const front = await startProxyFront(matchFor(service, '/'));
servers.push(front);
const response = await fetch(`${front.url}/my-service`);
expect(response.status).toBe(502);
});
it('responds 504 when the backend accepts the connection but never replies', async () => {
const backend = createHttpServer(() => {
// intentionally never respond
});
await new Promise<void>((resolve) => backend.listen(0, '127.0.0.1', resolve));
const address = backend.address();
if (address === null || typeof address === 'string') throw new Error('expected a TCP address');
const service: RegisteredService = {
baseUrl: `http://127.0.0.1:${address.port}`,
endpoints: [],
expiresAt: Date.now() + 60_000,
};
const front = await startProxyFront(matchFor(service, '/'), 200);
servers.push(front);
const response = await fetch(`${front.url}/my-service`);
expect(response.status).toBe(504);
await new Promise<void>((resolve, reject) => backend.close((e) => (e ? reject(e) : resolve())));
});
});

85
packages/server/service-discovery/src/aggregator/proxy-request.ts Normal file
View file

@ -0,0 +1,85 @@
import type { Agent as HttpAgent, IncomingMessage, ServerResponse } from 'node:http';
import { request as httpRequest } from 'node:http';
import type { Agent as HttpsAgent } from 'node:https';
import { request as httpsRequest } from 'node:https';
import { filterHopByHop } from './http-utils';
import type { RouteMatch } from './route-matching';
const LOG_PREFIX = '[lightweight-sdk:service-discovery:aggregator]';
/** Options controlling how {@link proxyRequest} relays a request to its matched backend. */
export interface ProxyOptions {
/** Socket inactivity timeout for the backend request, in milliseconds. Not a total duration cap. */
timeoutMs: number;
/** Keep-alive agent to use for `http:` targets. */
httpAgent: HttpAgent;
/** Keep-alive agent to use for `https:` targets. */
httpsAgent: HttpsAgent;
}
/**
* Relays `req` to the backend service identified by `match`, streaming the
* request and response bodies through with `node:http`/`node:https` — no
* buffering, no third-party proxy dependency. Responds directly on `res`;
* does not return a value the caller needs to act on.
* @param req the incoming request to relay
* @param res the response to write the backend's reply to
* @param match the route match identifying the backend to relay to
* @param options proxy behaviour options
*/
export function proxyRequest(
req: IncomingMessage,
res: ServerResponse,
match: RouteMatch,
options: ProxyOptions,
): void {
const search = req.url?.includes('?') ? req.url.slice(req.url.indexOf('?')) : '';
const targetUrl = new URL(`${match.remainderPath}${search}`, match.service.baseUrl);
const isHttps = targetUrl.protocol === 'https:';
const transportRequest = isHttps ? httpsRequest : httpRequest;
let timedOut = false;
const proxyReq = transportRequest(
{
hostname: targetUrl.hostname,
port: targetUrl.port,
path: `${targetUrl.pathname}${targetUrl.search}`,
method: req.method,
headers: { ...filterHopByHop(req.headers), host: targetUrl.host },
timeout: options.timeoutMs,
agent: isHttps ? options.httpsAgent : options.httpAgent,
},
(proxyRes) => {
res.writeHead(proxyRes.statusCode ?? 502, filterHopByHop(proxyRes.headers));
proxyRes.pipe(res);
},
);
proxyReq.on('timeout', () => {
timedOut = true;
proxyReq.destroy();
});
proxyReq.on('error', (error) => {
console.error(`${LOG_PREFIX} relay to "${targetUrl.origin}" failed`, error);
if (!res.headersSent) {
res.writeHead(timedOut ? 504 : 502, { 'content-type': 'application/json' });
res.end(JSON.stringify({ error: timedOut ? 'gateway timeout' : 'bad gateway' }));
} else {
res.destroy();
}
});
// Listening on `res` (not `req`) matters: `req`'s 'close' fires as soon as
// the incoming request body has been fully read — immediately for a
// bodyless GET, well before the backend has replied — which would abort
// the outbound request we just sent. `res` only closes early if the
// original caller actually disconnects before the response completes.
res.on('close', () => {
if (!res.writableEnded) {
proxyReq.destroy();
}
});
req.pipe(proxyReq);
}

11
packages/server/service-discovery/src/aggregator/registered-service.ts Normal file
View file

@ -0,0 +1,11 @@
import type { ServiceEndpoint } from '../types';
/** Internal record kept by {@link ServiceAggregator} for each registered service. */
export interface RegisteredService {
/** URL of base which the aggregator proxies matched requests to. */
baseUrl: string;
/** Endpoints this service declared at registration time. */
endpoints: ServiceEndpoint[];
/** Epoch milliseconds at which this registration expires without a heartbeat. */
expiresAt: number;
}

72
packages/server/service-discovery/src/aggregator/route-matching.test.ts Normal file
View file

@ -0,0 +1,72 @@
import { describe, expect, it } from 'vitest';
import type { RegisteredService } from './registered-service';
import { matchRoute } from './route-matching';
function registryWith(entries: Record<string, RegisteredService>): Map<string, RegisteredService> {
return new Map(Object.entries(entries));
}
const activeService: RegisteredService = {
baseUrl: 'http://127.0.0.1:9999',
endpoints: [{ method: 'GET', path: '/users' }],
expiresAt: Date.now() + 60_000,
};
describe('matchRoute', () => {
it('matches a registered endpoint under the service-name prefix', () => {
const registry = registryWith({ 'my-service': activeService });
const match = matchRoute(registry, 'GET', '/my-service/users');
expect(match).toEqual({
serviceName: 'my-service',
service: activeService,
remainderPath: '/users',
});
});
it('returns undefined for an unknown service', () => {
const registry = registryWith({ 'my-service': activeService });
expect(matchRoute(registry, 'GET', '/other-service/users')).toBeUndefined();
});
it('returns undefined when the method does not match', () => {
const registry = registryWith({ 'my-service': activeService });
expect(matchRoute(registry, 'POST', '/my-service/users')).toBeUndefined();
});
it('returns undefined when the path is not declared', () => {
const registry = registryWith({ 'my-service': activeService });
expect(matchRoute(registry, 'GET', '/my-service/unknown')).toBeUndefined();
});
it('returns undefined for an expired registration, even if still present', () => {
const expired: RegisteredService = { ...activeService, expiresAt: Date.now() - 1 };
const registry = registryWith({ 'my-service': expired });
expect(matchRoute(registry, 'GET', '/my-service/users')).toBeUndefined();
});
it('returns undefined for the root path', () => {
const registry = registryWith({ 'my-service': activeService });
expect(matchRoute(registry, 'GET', '/')).toBeUndefined();
});
it('matches the service root endpoint ("/") correctly', () => {
const rootService: RegisteredService = {
...activeService,
endpoints: [{ method: 'GET', path: '/' }],
};
const registry = registryWith({ 'my-service': rootService });
expect(matchRoute(registry, 'GET', '/my-service')).toEqual({
serviceName: 'my-service',
service: rootService,
remainderPath: '/',
});
});
});

45
packages/server/service-discovery/src/aggregator/route-matching.ts Normal file
View file

@ -0,0 +1,45 @@
import type { RegisteredService } from './registered-service';
/** Result of a successful {@link matchRoute} call. */
export interface RouteMatch {
/** Name of the service the incoming request was routed to. */
serviceName: string;
/** The matched service's registration record. */
service: RegisteredService;
/** Path to forward to the service, with its name prefix stripped (e.g. "/users"). */
remainderPath: string;
}
/**
* Matches an incoming request against the registry, using a "service name as
* first path segment" scheme: `GET /my-service/users` matches a service
* registered as `my-service` with an endpoint `GET /users`. Unknown
* services, expired registrations, and known services without a matching
* endpoint all resolve to `undefined` alike — callers surface this as a 404.
* @param registry the current service registry, keyed by service name
* @param method HTTP method of the incoming request
* @param pathname pathname of the incoming request (no query string)
* @returns the matched route, or `undefined` if nothing matches
*/
export function matchRoute(
registry: ReadonlyMap<string, RegisteredService>,
method: string,
pathname: string,
): RouteMatch | undefined {
const segments = pathname.split('/').filter((segment) => segment.length > 0);
if (segments.length === 0) {
return undefined;
}
const [serviceName, ...rest] = segments;
const service = registry.get(serviceName);
if (service === undefined || service.expiresAt <= Date.now()) {
return undefined;
}
const remainderPath = `/${rest.join('/')}`;
const matched = service.endpoints.some(
(endpoint) => endpoint.method === method && endpoint.path === remainderPath,
);
return matched ? { serviceName, service, remainderPath } : undefined;
}

144
packages/server/service-discovery/src/aggregator/service-aggregator.test.ts Normal file
View file

@ -0,0 +1,144 @@
import type { IncomingMessage, ServerResponse } from 'node:http';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import type { EphemeralServer } from '../test-support/ephemeral-http-server';
import { startEphemeralServer } from '../test-support/ephemeral-http-server';
import { ServiceAggregator } from './service-aggregator';
const SECRET = 'shh';
const PAYLOAD = {
serviceName: 'my-service',
baseUrl: 'http://127.0.0.1:9999',
endpoints: [{ method: 'GET', path: '/users' }],
};
async function startControlPlane(aggregator: ServiceAggregator): Promise<EphemeralServer> {
return startEphemeralServer((req: IncomingMessage, res: ServerResponse) => {
if (req.url === '/register') {
void aggregator.registrationHandler(req, res);
return;
}
if (req.url === '/deregister') {
void aggregator.deregistrationHandler(req, res);
return;
}
void aggregator.proxyHandler(req, res);
});
}
// `secret: null` (as opposed to the omitted/default case) means "send no
// Authorization header at all" — using `undefined` for that would silently
// fall back to the default parameter instead of actually omitting it.
function register(url: string, body: unknown, secret: string | null = SECRET) {
return fetch(`${url}/register`, {
method: 'POST',
headers: {
'content-type': 'application/json',
...(secret === null ? {} : { authorization: `Bearer ${secret}` }),
},
body: JSON.stringify(body),
});
}
function deregister(url: string, serviceName: string, secret: string | null = SECRET) {
return fetch(`${url}/deregister`, {
method: 'POST',
headers: {
'content-type': 'application/json',
...(secret === null ? {} : { authorization: `Bearer ${secret}` }),
},
body: JSON.stringify({ serviceName }),
});
}
describe('ServiceAggregator', () => {
let aggregator: ServiceAggregator;
let front: EphemeralServer;
beforeEach(async () => {
aggregator = new ServiceAggregator({ sharedSecret: SECRET, ttlMs: 1000, sweepIntervalMs: 500 });
front = await startControlPlane(aggregator);
});
afterEach(async () => {
aggregator.stop();
await front.close();
});
it('registers a valid payload and lists it', async () => {
const response = await register(front.url, PAYLOAD);
expect(response.status).toBe(200);
expect(await response.json()).toEqual({ ok: true, expiresInMs: 1000 });
expect(aggregator.getRegisteredServiceNames()).toEqual(['my-service']);
});
it('rejects registration with a missing secret', async () => {
const response = await register(front.url, PAYLOAD, null);
expect(response.status).toBe(401);
});
it('rejects registration with a wrong secret', async () => {
const response = await register(front.url, PAYLOAD, 'wrong');
expect(response.status).toBe(401);
});
it('rejects a malformed payload with 400', async () => {
const response = await register(front.url, { serviceName: 'x' }); // missing baseUrl/endpoints
expect(response.status).toBe(400);
});
it('treats re-registration under the same name as an upsert, not a duplicate error', async () => {
await register(front.url, PAYLOAD);
const second = await register(front.url, PAYLOAD);
expect(second.status).toBe(200);
expect(aggregator.getRegisteredServiceNames()).toEqual(['my-service']);
});
it('renews the TTL on heartbeat, surviving past the original expiry', async () => {
vi.useFakeTimers();
try {
await register(front.url, PAYLOAD);
await vi.advanceTimersByTimeAsync(900); // close to the 1000ms TTL
await register(front.url, PAYLOAD); // heartbeat renews it
await vi.advanceTimersByTimeAsync(600); // past the *original* expiry, not the renewed one
expect(aggregator.getRegisteredServiceNames()).toEqual(['my-service']);
} finally {
vi.useRealTimers();
}
});
it('evicts an expired registration once the sweep runs', async () => {
vi.useFakeTimers();
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {});
try {
aggregator.start();
await register(front.url, PAYLOAD);
await vi.advanceTimersByTimeAsync(1600); // past ttlMs (1000) and one sweep tick (500)
expect(aggregator.getRegisteredServiceNames()).toEqual([]);
expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining('evicted'));
} finally {
vi.useRealTimers();
}
});
it('deregisters a service immediately, before TTL expiry', async () => {
await register(front.url, PAYLOAD);
const response = await deregister(front.url, 'my-service');
expect(response.status).toBe(204);
expect(aggregator.getRegisteredServiceNames()).toEqual([]);
});
it('treats deregistering an unknown service as a no-op', async () => {
const response = await deregister(front.url, 'unknown-service');
expect(response.status).toBe(204);
});
it('rejects deregistration without a valid secret, same as registration', async () => {
const response = await deregister(front.url, 'my-service', 'wrong');
expect(response.status).toBe(401);
});
});

185
packages/server/service-discovery/src/aggregator/service-aggregator.ts Normal file
View file

@ -0,0 +1,185 @@
import type { IncomingMessage, ServerResponse } from 'node:http';
import { Agent as HttpAgent } from 'node:http';
import { Agent as HttpsAgent } from 'node:https';
import type { RegistrationAck } from '../types';
import { readJsonBody, timingSafeEqualStrings } from './http-utils';
import { proxyRequest } from './proxy-request';
import type { RegisteredService } from './registered-service';
import { matchRoute } from './route-matching';
import {
validateDeregistrationPayload,
validateRegistrationPayload,
} from './validate-registration';
const LOG_PREFIX = '[lightweight-sdk:service-discovery:aggregator]';
const DEFAULT_TTL_MS = 45_000;
const DEFAULT_SWEEP_INTERVAL_MS = 10_000;
const DEFAULT_PROXY_TIMEOUT_MS = 10_000;
const DEFAULT_MAX_REGISTRATION_BODY_BYTES = 65_536;
/** Configuration options for a {@link ServiceAggregator}. */
export interface ServiceAggregatorOptions {
/** Secret that registering services must present via `Authorization: Bearer <secret>`. */
sharedSecret: string;
/** How long a registration stays valid without a heartbeat, in milliseconds.
* @default 45000 */
ttlMs?: number;
/** How often to sweep and evict expired registrations, in milliseconds.
* @default 10000 */
sweepIntervalMs?: number;
/** Socket inactivity timeout for proxied requests to backends, in milliseconds.
* @default 10000 */
proxyTimeoutMs?: number;
/** Maximum accepted size for a registration/deregistration body, in bytes.
* @default 65536 */
maxRegistrationBodyBytes?: number;
}
/**
* Reverse-proxy API aggregator: services register themselves (and their
* endpoints) at runtime, and the aggregator relays matching incoming
* requests to them. Exposes plain Node request handlers — mount them into
* an existing Express/Fastify/`node:http` server, this class never listens
* on a port itself.
*/
export class ServiceAggregator {
private readonly options: ServiceAggregatorOptions;
private readonly registry = new Map<string, RegisteredService>();
private readonly httpAgent = new HttpAgent({ keepAlive: true });
private readonly httpsAgent = new HttpsAgent({ keepAlive: true });
private sweepTimer: NodeJS.Timeout | undefined;
public constructor(options: ServiceAggregatorOptions) {
this.options = options;
}
/** Handles a service registration or heartbeat request. Mount on whatever path you choose (e.g. `POST /discovery/register`). */
public readonly registrationHandler = async (
req: IncomingMessage,
res: ServerResponse,
): Promise<void> => {
if (!this.isAuthorized(req)) {
this.respondJson(res, 401, { error: 'unauthorized' });
return;
}
try {
const body = await readJsonBody(
req,
this.options.maxRegistrationBodyBytes ?? DEFAULT_MAX_REGISTRATION_BODY_BYTES,
);
const payload = validateRegistrationPayload(body);
const ttlMs = this.options.ttlMs ?? DEFAULT_TTL_MS;
this.registry.set(payload.serviceName, {
baseUrl: payload.baseUrl,
endpoints: payload.endpoints,
expiresAt: Date.now() + ttlMs,
});
console.debug(
`${LOG_PREFIX} registered "${payload.serviceName}" (${payload.endpoints.length} endpoint(s))`,
);
this.respondJson(res, 200, { ok: true, expiresInMs: ttlMs } satisfies RegistrationAck);
} catch (error) {
console.error(`${LOG_PREFIX} registration failed`, error);
this.respondJson(res, 400, { error: 'invalid registration payload' });
}
};
/** Handles a best-effort service deregistration request. Mount on whatever path you choose (e.g. `POST /discovery/deregister`). */
public readonly deregistrationHandler = async (
req: IncomingMessage,
res: ServerResponse,
): Promise<void> => {
if (!this.isAuthorized(req)) {
this.respondJson(res, 401, { error: 'unauthorized' });
return;
}
try {
const body = await readJsonBody(
req,
this.options.maxRegistrationBodyBytes ?? DEFAULT_MAX_REGISTRATION_BODY_BYTES,
);
const payload = validateDeregistrationPayload(body);
if (this.registry.delete(payload.serviceName)) {
console.debug(`${LOG_PREFIX} deregistered "${payload.serviceName}"`);
}
res.writeHead(204);
res.end();
} catch (error) {
console.error(`${LOG_PREFIX} deregistration failed`, error);
this.respondJson(res, 400, { error: 'invalid deregistration payload' });
}
};
/** Handles a request that should be relayed to a registered service. Mount as a catch-all, after your own routes and the two handlers above. */
public readonly proxyHandler = async (
req: IncomingMessage,
res: ServerResponse,
): Promise<void> => {
const pathname = (req.url ?? '/').split('?')[0] ?? '/';
const match = matchRoute(this.registry, req.method ?? 'GET', pathname);
if (match === undefined) {
this.respondJson(res, 404, { error: 'no service registered for this route' });
return;
}
proxyRequest(req, res, match, {
timeoutMs: this.options.proxyTimeoutMs ?? DEFAULT_PROXY_TIMEOUT_MS,
httpAgent: this.httpAgent,
httpsAgent: this.httpsAgent,
});
};
/** Starts the periodic sweep that evicts expired registrations. */
public start(): void {
if (this.sweepTimer !== undefined) {
return;
}
this.sweepTimer = setInterval(
() => this.sweepExpired(),
this.options.sweepIntervalMs ?? DEFAULT_SWEEP_INTERVAL_MS,
);
this.sweepTimer.unref();
}
/** Stops the periodic sweep and releases the keep-alive connection pools. */
public stop(): void {
if (this.sweepTimer !== undefined) {
clearInterval(this.sweepTimer);
this.sweepTimer = undefined;
}
this.httpAgent.destroy();
this.httpsAgent.destroy();
}
/**
* Lists the names of every currently registered (non-expired) service —
* useful for introspection/health checks in the host app.
* @returns the registered service names
*/
public getRegisteredServiceNames(): string[] {
return [...this.registry.keys()];
}
private isAuthorized(req: IncomingMessage): boolean {
const header = req.headers.authorization;
if (header === undefined || !header.startsWith('Bearer ')) {
return false;
}
return timingSafeEqualStrings(header.slice('Bearer '.length), this.options.sharedSecret);
}
private respondJson(res: ServerResponse, status: number, body: unknown): void {
res.writeHead(status, { 'content-type': 'application/json' });
res.end(JSON.stringify(body));
}
private sweepExpired(): void {
const now = Date.now();
for (const [name, service] of this.registry) {
if (service.expiresAt <= now) {
this.registry.delete(name);
console.warn(`${LOG_PREFIX} evicted expired service "${name}"`);
}
}
}
}

64
packages/server/service-discovery/src/aggregator/validate-registration.test.ts Normal file
View file

@ -0,0 +1,64 @@
import { describe, expect, it } from 'vitest';
import {
validateDeregistrationPayload,
validateRegistrationPayload,
} from './validate-registration';
const validPayload = {
serviceName: 'my-service',
baseUrl: 'http://127.0.0.1:9999',
endpoints: [{ method: 'GET', path: '/users' }],
};
describe('validateRegistrationPayload', () => {
it('accepts a valid payload', () => {
expect(validateRegistrationPayload(validPayload)).toEqual(validPayload);
});
// Substring matching (not a RegExp): some messages contain regex
// metacharacters ("http(s)") that must be matched literally.
it.each([
['must be a JSON object', 'null payload', null],
// an array passes the typeof==='object' check; it then fails on the
// (undefined) destructured serviceName, same as an empty object would.
['"serviceName" must be a string', 'array payload', []],
[
'"serviceName" must be a string',
'missing serviceName',
{ ...validPayload, serviceName: undefined },
],
[
'Invalid service name',
'invalid serviceName casing',
{ ...validPayload, serviceName: 'Not-Kebab' },
],
['"baseUrl" must be a string', 'missing baseUrl', { ...validPayload, baseUrl: 42 }],
['valid http(s) URL', 'unparseable baseUrl', { ...validPayload, baseUrl: 'not a url' }],
['valid http(s) URL', 'non-http baseUrl', { ...validPayload, baseUrl: 'ftp://127.0.0.1' }],
['"endpoints" must be an array', 'non-array endpoints', { ...validPayload, endpoints: {} }],
[
'must be one of',
'invalid endpoint method',
{ ...validPayload, endpoints: [{ method: 'TRACE', path: '/x' }] },
],
[
'must start with "/"',
'endpoint path without leading slash',
{ ...validPayload, endpoints: [{ method: 'GET', path: 'x' }] },
],
])('rejects %s (%s)', (expectedMessage, _label, payload) => {
expect(() => validateRegistrationPayload(payload)).toThrow(expectedMessage);
});
});
describe('validateDeregistrationPayload', () => {
it('accepts a valid payload', () => {
expect(validateDeregistrationPayload({ serviceName: 'my-service' })).toEqual({
serviceName: 'my-service',
});
});
it('rejects a payload without a serviceName', () => {
expect(() => validateDeregistrationPayload({})).toThrow('"serviceName" must be a string');
});
});

94
packages/server/service-discovery/src/aggregator/validate-registration.ts Normal file
View file

@ -0,0 +1,94 @@
import { assertValidServiceName } from '../service-name';
import type {
DeregistrationPayload,
HttpMethod,
RegistrationPayload,
ServiceEndpoint,
} from '../types';
const HTTP_METHODS: ReadonlySet<string> = new Set<HttpMethod>([
'GET',
'POST',
'PUT',
'PATCH',
'DELETE',
'HEAD',
'OPTIONS',
]);
/**
* Validates and narrows an unknown parsed JSON value into a
* {@link RegistrationPayload}.
* @param payload the parsed JSON body to validate
* @returns the validated payload
* @throws {Error} with a descriptive message if `payload` does not match the expected shape
*/
export function validateRegistrationPayload(payload: unknown): RegistrationPayload {
if (typeof payload !== 'object' || payload === null) {
throw new Error('registration payload must be a JSON object');
}
const { serviceName, baseUrl, endpoints } = payload as Record<string, unknown>;
if (typeof serviceName !== 'string') {
throw new Error('"serviceName" must be a string');
}
assertValidServiceName(serviceName);
if (typeof baseUrl !== 'string') {
throw new Error('"baseUrl" must be a string');
}
const parsedBaseUrl = safeParseUrl(baseUrl);
if (
parsedBaseUrl === undefined ||
(parsedBaseUrl.protocol !== 'http:' && parsedBaseUrl.protocol !== 'https:')
) {
throw new Error('"baseUrl" must be a valid http(s) URL');
}
if (!Array.isArray(endpoints)) {
throw new Error('"endpoints" must be an array');
}
const validatedEndpoints = endpoints.map((endpoint, index) => validateEndpoint(endpoint, index));
return { serviceName, baseUrl, endpoints: validatedEndpoints };
}
/**
* Validates and narrows an unknown parsed JSON value into a
* {@link DeregistrationPayload}.
* @param payload the parsed JSON body to validate
* @returns the validated payload
* @throws {Error} with a descriptive message if `payload` does not match the expected shape
*/
export function validateDeregistrationPayload(payload: unknown): DeregistrationPayload {
if (typeof payload !== 'object' || payload === null) {
throw new Error('deregistration payload must be a JSON object');
}
const { serviceName } = payload as Record<string, unknown>;
if (typeof serviceName !== 'string') {
throw new Error('"serviceName" must be a string');
}
return { serviceName };
}
function validateEndpoint(endpoint: unknown, index: number): ServiceEndpoint {
if (typeof endpoint !== 'object' || endpoint === null) {
throw new Error(`endpoints[${index}] must be an object`);
}
const { method, path } = endpoint as Record<string, unknown>;
if (typeof method !== 'string' || !HTTP_METHODS.has(method)) {
throw new Error(`endpoints[${index}].method must be one of ${[...HTTP_METHODS].join(', ')}`);
}
if (typeof path !== 'string' || !path.startsWith('/')) {
throw new Error(`endpoints[${index}].path must start with "/"`);
}
return { method: method as HttpMethod, path };
}
function safeParseUrl(value: string): URL | undefined {
try {
return new URL(value);
} catch {
return undefined;
}
}

150
packages/server/service-discovery/src/client/discovery-client.test.ts Normal file
View file

@ -0,0 +1,150 @@
import type { IncomingMessage, ServerResponse } from 'node:http';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import type { EphemeralServer } from '../test-support/ephemeral-http-server';
import { startEphemeralServer } from '../test-support/ephemeral-http-server';
import type { RegistrationPayload } from '../types';
import type { DiscoveryClientOptions } from './discovery-client';
import { DiscoveryClient } from './discovery-client';
interface RecordedRequest {
method: string | undefined;
url: string | undefined;
authorization: string | undefined;
body: unknown;
}
async function readBody(req: IncomingMessage): Promise<unknown> {
const chunks: Buffer[] = [];
for await (const chunk of req) {
chunks.push(chunk as Buffer);
}
const raw = Buffer.concat(chunks).toString('utf8');
return raw.length > 0 ? JSON.parse(raw) : undefined;
}
/** Fake aggregator recording every request it receives, responding with `status`. */
async function startRecordingAggregator(
status = 200,
): Promise<EphemeralServer & { requests: RecordedRequest[] }> {
const requests: RecordedRequest[] = [];
const server = await startEphemeralServer((req: IncomingMessage, res: ServerResponse) => {
void readBody(req).then((body) => {
requests.push({
method: req.method,
url: req.url,
authorization: req.headers.authorization,
body,
});
res.writeHead(status, { 'content-type': 'application/json' });
res.end('{}');
});
});
return Object.assign(server, { requests });
}
describe('DiscoveryClient', () => {
let aggregator: Awaited<ReturnType<typeof startRecordingAggregator>>;
let client: DiscoveryClient | undefined;
beforeEach(async () => {
vi.useFakeTimers();
aggregator = await startRecordingAggregator();
});
afterEach(async () => {
await client?.stop();
client = undefined;
await aggregator.close();
vi.useRealTimers();
});
function createClient(overrides: Partial<DiscoveryClientOptions> = {}) {
client = new DiscoveryClient({
aggregatorUrl: aggregator.url,
serviceName: 'my-service',
baseUrl: 'http://127.0.0.1:9999',
endpoints: [{ method: 'GET', path: '/users' }],
sharedSecret: 'shh',
heartbeatIntervalMs: 1000,
...overrides,
});
return client;
}
it('sends the initial registration with the payload and the Authorization header', async () => {
await createClient().start();
expect(aggregator.requests).toHaveLength(1);
expect(aggregator.requests[0]?.method).toBe('POST');
expect(aggregator.requests[0]?.url).toBe('/discovery/register');
expect(aggregator.requests[0]?.authorization).toBe('Bearer shh');
expect(aggregator.requests[0]?.body).toEqual({
serviceName: 'my-service',
baseUrl: 'http://127.0.0.1:9999',
endpoints: [{ method: 'GET', path: '/users' }],
} satisfies RegistrationPayload);
});
it('sends a heartbeat on every interval tick', async () => {
await createClient().start();
await vi.advanceTimersByTimeAsync(1000);
await vi.advanceTimersByTimeAsync(1000);
expect(aggregator.requests).toHaveLength(3); // initial + 2 heartbeats
});
it('logs a failed heartbeat and keeps heartbeating afterwards', async () => {
await aggregator.close();
aggregator = await startRecordingAggregator(500);
const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {});
await createClient().start();
expect(errorSpy).toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1000);
expect(aggregator.requests).toHaveLength(2); // still ticking despite the 500s
});
it('stops sending heartbeats after stop()', async () => {
const c = createClient();
await c.start();
await c.stop();
client = undefined; // already stopped, afterEach should not double-stop
const countAfterStop = aggregator.requests.length;
await vi.advanceTimersByTimeAsync(5000);
expect(aggregator.requests).toHaveLength(countAfterStop);
});
it('sends a best-effort deregistration on stop()', async () => {
const c = createClient();
await c.start();
await c.stop();
client = undefined;
const deregistration = aggregator.requests.find((r) => r.url === '/discovery/deregister');
expect(deregistration).toBeDefined();
expect(deregistration?.body).toEqual({ serviceName: 'my-service' });
});
it('never rejects on stop() even if deregistration fails', async () => {
const c = createClient({ aggregatorUrl: 'http://127.0.0.1:1' }); // nothing listens here
await expect(c.stop()).resolves.toBeUndefined();
client = undefined;
});
it('does not throw when stop() is called without a prior start()', async () => {
const c = createClient();
await expect(c.stop()).resolves.toBeUndefined();
client = undefined;
});
it('does not start a second heartbeat loop when start() is called twice', async () => {
const c = createClient();
await c.start();
await c.start();
await vi.advanceTimersByTimeAsync(1000);
expect(aggregator.requests).toHaveLength(2); // initial + 1 heartbeat, not 3
});
});

151
packages/server/service-discovery/src/client/discovery-client.ts Normal file
View file

@ -0,0 +1,151 @@
import { assertValidServiceName } from '../service-name';
import type { DeregistrationPayload, RegistrationPayload, ServiceEndpoint } from '../types';
const LOG_PREFIX = '[lightweight-sdk:service-discovery:client]';
/** Configuration options for a {@link DiscoveryClient}. */
export interface DiscoveryClientOptions {
/** Base URL of the aggregator to register with (e.g. "http://localhost:8080"). */
aggregatorUrl: string;
/** Unique, stable name for this service — also becomes its public routing prefix. */
serviceName: string;
/** Base URL the aggregator will proxy matched requests to. */
baseUrl: string;
/** Endpoints this service exposes. */
endpoints: ServiceEndpoint[];
/** Secret presented to the aggregator to authenticate registration. */
sharedSecret: string;
/** How often to re-send the registration as a heartbeat, in milliseconds.
* @default 15000 */
heartbeatIntervalMs?: number;
/** Timeout for each registration/heartbeat/deregistration request, in milliseconds.
* @default 5000 */
requestTimeoutMs?: number;
/** Path on the aggregator that handles registration.
* @default '/discovery/register' */
registerPath?: string;
/** Path on the aggregator that handles deregistration.
* @default '/discovery/deregister' */
deregisterPath?: string;
}
const DEFAULT_HEARTBEAT_INTERVAL_MS = 15_000;
const DEFAULT_REQUEST_TIMEOUT_MS = 5_000;
const DEFAULT_REGISTER_PATH = '/discovery/register';
const DEFAULT_DEREGISTER_PATH = '/discovery/deregister';
/**
* Registers a service with a ServiceAggregator (`@lightweight-sdk/service-discovery/aggregator`) and keeps the
* registration alive with a periodic heartbeat. The heartbeat is also the
* retry mechanism: a failed registration or heartbeat is logged and left to
* the next scheduled tick, rather than retried immediately — this covers
* both "aggregator not up yet" and transient outages without extra logic.
*/
export class DiscoveryClient {
private readonly options: DiscoveryClientOptions;
private readonly registerUrl: string;
private readonly deregisterUrl: string;
private heartbeatTimer: NodeJS.Timeout | undefined;
/**
* @param options configuration for this client
* @throws {Error} if `options.serviceName` is not valid kebab-case
*/
public constructor(options: DiscoveryClientOptions) {
assertValidServiceName(options.serviceName);
this.options = options;
this.registerUrl = new URL(
options.registerPath ?? DEFAULT_REGISTER_PATH,
options.aggregatorUrl,
).toString();
this.deregisterUrl = new URL(
options.deregisterPath ?? DEFAULT_DEREGISTER_PATH,
options.aggregatorUrl,
).toString();
}
/**
* Sends the initial registration and starts the periodic heartbeat. Never
* rejects: a failed initial registration is logged and retried on the
* next heartbeat tick, same as any other heartbeat failure.
*/
public async start(): Promise<void> {
if (this.heartbeatTimer !== undefined) {
console.warn(`${LOG_PREFIX} start() called while already started — ignored`);
return;
}
await this.sendHeartbeat();
this.heartbeatTimer = setInterval(() => {
this.sendHeartbeat().catch((error: unknown) => {
console.error(`${LOG_PREFIX} unexpected heartbeat error`, error);
});
}, this.options.heartbeatIntervalMs ?? DEFAULT_HEARTBEAT_INTERVAL_MS);
this.heartbeatTimer.unref();
}
/**
* Stops the heartbeat and attempts a best-effort deregistration. Never
* throws: a failed deregistration is logged as a warning (the aggregator
* will evict this service via TTL regardless), so callers can always
* `await stop()` during a graceful shutdown without extra error handling.
*/
public async stop(): Promise<void> {
if (this.heartbeatTimer !== undefined) {
clearInterval(this.heartbeatTimer);
this.heartbeatTimer = undefined;
}
try {
await this.sendDeregistration();
} catch (error) {
console.warn(
`${LOG_PREFIX} best-effort deregistration failed — aggregator will evict via TTL`,
error,
);
}
}
private async sendHeartbeat(): Promise<void> {
try {
const response = await fetch(this.registerUrl, {
method: 'POST',
headers: {
'content-type': 'application/json',
authorization: `Bearer ${this.options.sharedSecret}`,
},
body: JSON.stringify(this.buildPayload()),
signal: AbortSignal.timeout(this.options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS),
});
if (!response.ok) {
throw new Error(`aggregator responded ${response.status}`);
}
console.debug(`${LOG_PREFIX} heartbeat ok for "${this.options.serviceName}"`);
} catch (error) {
console.error(`${LOG_PREFIX} heartbeat failed for "${this.options.serviceName}"`, error);
}
}
private async sendDeregistration(): Promise<void> {
const response = await fetch(this.deregisterUrl, {
method: 'POST',
headers: {
'content-type': 'application/json',
authorization: `Bearer ${this.options.sharedSecret}`,
},
body: JSON.stringify({
serviceName: this.options.serviceName,
} satisfies DeregistrationPayload),
signal: AbortSignal.timeout(this.options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS),
});
if (!response.ok) {
throw new Error(`aggregator responded ${response.status}`);
}
}
private buildPayload(): RegistrationPayload {
return {
serviceName: this.options.serviceName,
baseUrl: this.options.baseUrl,
endpoints: this.options.endpoints,
};
}
}

3
packages/server/service-discovery/src/client/index.ts Normal file
View file

@ -0,0 +1,3 @@
export type { HttpMethod, ServiceEndpoint } from '../types';
export type { DiscoveryClientOptions } from './discovery-client';
export { DiscoveryClient } from './discovery-client';

7
packages/server/service-discovery/src/index.ts Normal file
View file

@ -0,0 +1,7 @@
export type {
DeregistrationPayload,
HttpMethod,
RegistrationAck,
RegistrationPayload,
ServiceEndpoint,
} from './types';

17
packages/server/service-discovery/src/service-name.ts Normal file
View file

@ -0,0 +1,17 @@
const SERVICE_NAME_PATTERN = /^[a-z][a-z0-9-]*[a-z0-9]$/;
/**
* Validates that `serviceName` is a valid service identifier (kebab-case),
* safe to use as a URL path segment for routing. Shared between the client
* (fail-fast at construction) and the aggregator (registration payload
* validation).
* @param serviceName the candidate service name
* @throws {Error} if `serviceName` does not match the expected kebab-case format
*/
export function assertValidServiceName(serviceName: string): void {
if (!SERVICE_NAME_PATTERN.test(serviceName)) {
throw new Error(
`Invalid service name "${serviceName}": expected kebab-case (e.g. "my-service").`,
);
}
}

39
packages/server/service-discovery/src/test-support/ephemeral-http-server.ts Normal file
View file

@ -0,0 +1,39 @@
import type { RequestListener } from 'node:http';
import { createServer } from 'node:http';
/** A running ephemeral HTTP server started by {@link startEphemeralServer}. */
export interface EphemeralServer {
/** Base URL of the running server (e.g. "http://127.0.0.1:54321"). */
url: string;
/** Port the server is listening on. */
port: number;
/** Stops the server and waits until the underlying socket is closed. */
close: () => Promise<void>;
}
/**
* Starts a real `node:http` server on a random free port. Used across the
* test suite as a stand-in aggregator or backend service, instead of
* mocking `fetch`/`node:http` — this exercises the real proxy/HTTP
* plumbing rather than a mocked approximation of it.
* @param handler request handler for the ephemeral server
* @returns the running server's URL/port and a `close` function
*/
export async function startEphemeralServer(handler: RequestListener): Promise<EphemeralServer> {
const server = createServer(handler);
await new Promise<void>((resolve) => {
server.listen(0, '127.0.0.1', resolve);
});
const address = server.address();
if (address === null || typeof address === 'string') {
throw new Error('expected the ephemeral server to bind to a TCP address');
}
return {
url: `http://127.0.0.1:${address.port}`,
port: address.port,
close: () =>
new Promise<void>((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
}),
};
}

37
packages/server/service-discovery/src/types.ts Normal file
View file

@ -0,0 +1,37 @@
/** Méthodes HTTP qu'un endpoint de service enregistré peut exposer. */
export type HttpMethod = 'GET' | 'POST' | 'PUT' | 'PATCH' | 'DELETE' | 'HEAD' | 'OPTIONS';
/** Un endpoint exposé par un service, relatif à son `baseUrl`. */
export interface ServiceEndpoint {
/** Méthode HTTP à laquelle cet endpoint répond. */
method: HttpMethod;
/** Chemin relatif au `baseUrl` du service (ex. "/users"). Doit commencer par "/", sans query string. */
path: string;
}
/**
* Payload envoyé par un service pour s'enregistrer ou renouveler son
* enregistrement (heartbeat) — c'est la même opération d'upsert des deux côtés.
*/
export interface RegistrationPayload {
/** Nom unique et stable du service — sert aussi de préfixe de routage public. */
serviceName: string;
/** URL de base vers laquelle l'agrégateur relaiera les requêtes matchées. */
baseUrl: string;
/** Endpoints exposés par ce service. */
endpoints: ServiceEndpoint[];
}
/** Réponse de l'agrégateur sur un enregistrement ou un heartbeat accepté. */
export interface RegistrationAck {
/** Toujours `true` — les échecs utilisent un statut HTTP d'erreur, jamais ce corps. */
ok: true;
/** Millisecondes avant expiration si aucun heartbeat suivant n'est reçu — informatif. */
expiresInMs: number;
}
/** Payload envoyé pour une désinscription explicite (best-effort, à l'arrêt d'un service). */
export interface DeregistrationPayload {
/** Nom du service à désinscrire, tel que fourni lors de l'enregistrement. */
serviceName: string;
}

9
packages/server/service-discovery/tsconfig.json Normal file
View file

@ -0,0 +1,9 @@
{
"extends": "../../../tsconfig.base.json",
"compilerOptions": {
"outDir": "./dist",
"rootDir": "./src"
},
"include": ["src/**/*"],
"exclude": ["src/**/*.test.ts"]
}

12
pnpm-lock.yaml
View file

@ -223,6 +223,18 @@ importers:
specifier: 'catalog:' specifier: 'catalog:'
version: 2.1.9(@types/node@22.20.3) version: 2.1.9(@types/node@22.20.3)
packages/server/service-discovery:
devDependencies:
'@types/node':
specifier: 'catalog:'
version: 22.20.3
typescript:
specifier: 'catalog:'
version: 5.9.3
vitest:
specifier: 'catalog:'
version: 2.1.9(@types/node@22.20.3)
packages: packages:
'@babel/runtime@7.29.7': '@babel/runtime@7.29.7':

20
scripts/create-package.mjs
View file

@ -123,11 +123,23 @@ rootTsconfig.references ??= [];
rootTsconfig.references.push({ path: `./packages/${category}/${name}` }); rootTsconfig.references.push({ path: `./packages/${category}/${name}` });
writeFileSync(rootTsconfigPath, `${JSON.stringify(rootTsconfig, null, '\t')}\n`); writeFileSync(rootTsconfigPath, `${JSON.stringify(rootTsconfig, null, '\t')}\n`);
// Ajoute une ligne au registre des packages // Ajoute une ligne dans le tableau du registre des packages (pas en fin de
// fichier : insérée juste après la dernière ligne existante du tableau,
// même si d'autres sections suivent plus bas dans le fichier).
const packagesSpecPath = join(ROOT, 'spec', 'packages.md'); const packagesSpecPath = join(ROOT, 'spec', 'packages.md');
const packagesSpec = readFileSync(packagesSpecPath, 'utf8'); const packagesSpecLines = readFileSync(packagesSpecPath, 'utf8').split('\n');
const newRow = `| \`${npmName}\` | ${category} | 0.0.1 | brouillon |\n`; const newRow = `| \`${npmName}\` | ${category} | 0.0.1 | brouillon |`;
writeFileSync(packagesSpecPath, packagesSpec.trimEnd() + '\n' + newRow); const separatorIndex = packagesSpecLines.findIndex((line) => line.startsWith('|---'));
if (separatorIndex === -1) {
packagesSpecLines.push(newRow);
} else {
let insertAt = separatorIndex + 1;
while (insertAt < packagesSpecLines.length && packagesSpecLines[insertAt].startsWith('|')) {
insertAt += 1;
}
packagesSpecLines.splice(insertAt, 0, newRow);
}
writeFileSync(packagesSpecPath, packagesSpecLines.join('\n'));
console.log(`Package ${npmName} créé dans packages/${category}/${name}/.`); console.log(`Package ${npmName} créé dans packages/${category}/${name}/.`);

15
spec/architecture.md
View file

@ -34,6 +34,21 @@ Un nouveau package se crée avec `pnpm create-package <client|common|server> <no
`tsconfig.base.json` force `"useDefineForClassFields": false`, bien que la cible soit `ES2022` (où ce flag vaut `true` par défaut). Sans ça, un champ de classe déclaré sans initialiseur (`logger!: LoggerService;`) est réinitialisé à `undefined` **sur l'instance** au moment de la construction, ce qui masque silencieusement tout accesseur posé sur le prototype par un décorateur de propriété (`@Service` de `@lightweight-sdk/service-registry`, `@LinkToDom` de `web-component`). C'est une interaction connue entre `useDefineForClassFields` et les décorateurs legacy (`experimentalDecorators`) — désactiver ce flag restaure la sémantique classique où un champ sans initialiseur n'émet aucun code, laissant l'accesseur du prototype pleinement fonctionner. `tsconfig.base.json` force `"useDefineForClassFields": false`, bien que la cible soit `ES2022` (où ce flag vaut `true` par défaut). Sans ça, un champ de classe déclaré sans initialiseur (`logger!: LoggerService;`) est réinitialisé à `undefined` **sur l'instance** au moment de la construction, ce qui masque silencieusement tout accesseur posé sur le prototype par un décorateur de propriété (`@Service` de `@lightweight-sdk/service-registry`, `@LinkToDom` de `web-component`). C'est une interaction connue entre `useDefineForClassFields` et les décorateurs legacy (`experimentalDecorators`) — désactiver ce flag restaure la sémantique classique où un champ sans initialiseur n'émet aucun code, laissant l'accesseur du prototype pleinement fonctionner.
## Packages multi-points d'entrée
Un package peut exposer plusieurs points d'entrée séparés (ex. `@lightweight-sdk/service-discovery/client` et `.../aggregator`) quand son usage se scinde naturellement en deux surfaces qui ne doivent pas forcément être importées ensemble, tout en partageant un même protocole/types de base (une seule version à faire évoluer, pas de risque de dérive entre deux packages séparés). Convention établie avec `service-discovery`, à réutiliser pour de futurs packages `server/` dans le même cas :
- **Résolution sous CommonJS/`Node10`** : pas de champ `exports` moderne (ce mode de résolution ne le comprend pas). À la place, un répertoire stub à la racine du package par sous-chemin, contenant un mini `package.json` qui pointe vers le vrai fichier compilé — ex. `client/package.json` avec `{ "main": "../dist/client/index.js", "types": "../dist/client/index.d.ts" }`. Ces stubs doivent être listés dans `"files"` du `package.json` du package pour être effectivement publiés.
- **Un seul `tsconfig.json`** pour tout le package (pas un projet TS par point d'entrée) : `tsc -b` compile naturellement chaque `src/<sous-chemin>/index.ts` vers `dist/<sous-chemin>/index.js` en préservant l'arborescence.
- **Ré-exports nommés explicites obligatoires** dans chaque `index.ts` (jamais `export *`) : chaque point d'entrée doit ré-exporter les types du protocole partagé qu'il utilise dans sa surface publique, sinon la validation TypeDoc (bloquante en CI) échoue. `typedocOptions.entryPoints` dans le `package.json` du package doit lister les trois (ou plus) fichiers d'entrée.
## Pattern de handler framework-agnostique (packages `server/`)
Pour qu'un package `server/` reste utilisable quel que soit le framework HTTP de l'app hôte (le parc existant de l'auteur est déjà hétérogène : Express sur un projet, Fastify sur un autre), exposer la logique métier comme des handlers au format natif Node — `(req: IncomingMessage, res: ServerResponse) => void | Promise<void>` — plutôt que de dépendre d'Express, Fastify ou une lib de routing. L'app hôte les monte elle-même :
- **Express** : compatible direct, `req`/`res` Express *sont* des `IncomingMessage`/`ServerResponse`.
- **Fastify** : nécessite `reply.hijack()` avant d'appeler le handler avec `request.raw`/`reply.raw`, pour indiquer à Fastify de ne plus toucher `reply` ensuite.
- **`node:http` brut** : correspond exactement à la signature, aucun adaptateur nécessaire.
## Dépendances entre packages ## Dépendances entre packages
Les dépendances internes entre packages du monorepo utilisent le protocole `workspace:*` (résolu localement par pnpm, jamais via le registre distant). Le registre npm privé n'est utilisé que pour la publication et pour la consommation depuis des projets *externes* à ce monorepo. Les dépendances internes entre packages du monorepo utilisent le protocole `workspace:*` (résolu localement par pnpm, jamais via le registre distant). Le registre npm privé n'est utilisé que pour la publication et pour la consommation depuis des projets *externes* à ce monorepo.

4
spec/packages.md
View file

@ -3,6 +3,8 @@
| Package | Catégorie | Version | Statut | | Package | Catégorie | Version | Statut |
|---|---|---|---| |---|---|---|---|
| `@lightweight-sdk/web-component` | client | 0.0.2 | migré depuis `lightweight-sdk-web-component`, non modifié | | `@lightweight-sdk/web-component` | client | 0.0.2 | migré depuis `lightweight-sdk-web-component`, non modifié |
| `@lightweight-sdk/service-registry` | common | 0.0.1 | brouillon |
| `@lightweight-sdk/service-discovery` | server | 0.0.1 | brouillon |
## Ajouter un package ## Ajouter un package
@ -19,6 +21,7 @@ Le script scaffolde `package.json`, `tsconfig.json`, `src/index.ts` + un test, u
- `engines.node >= 22` - `engines.node >= 22`
- `publishConfig.registry` pointant vers le registre npm privé (`https://git.dev.kyuno.fr/api/packages/kyuno/npm/`) - `publishConfig.registry` pointant vers le registre npm privé (`https://git.dev.kyuno.fr/api/packages/kyuno/npm/`)
- `typedocOptions.entryPoints` (sauf exception documentée, voir ci-dessous) - `typedocOptions.entryPoints` (sauf exception documentée, voir ci-dessous)
- `devDependencies` explicites pour `typescript`/`vitest`/`@types/node` (`catalog:`), même si déjà hoistées depuis la racine — un package sans dépendance propre peut se retrouver sans entrée dans `pnpm-lock.yaml`, cassant `pnpm install --frozen-lockfile` en CI (voir `spec/ci-cd.md`)
## Observations sur `@lightweight-sdk/web-component` (non traitées, laissées telles quelles) ## Observations sur `@lightweight-sdk/web-component` (non traitées, laissées telles quelles)
@ -30,4 +33,3 @@ Le code source de ce package a été migré strictement à l'identique depuis l'
- Le style de formatage d'origine (4 espaces, guillemets doubles) diffère de la config Biome du monorepo (tabulations, guillemets simples) — le formatter Biome est désactivé pour ce package via le même `overrides`. - Le style de formatage d'origine (4 espaces, guillemets doubles) diffère de la config Biome du monorepo (tabulations, guillemets simples) — le formatter Biome est désactivé pour ce package via le même `overrides`.
- `REQUEST_DOM_LINK_KEY` semble ne plus être utilisé nulle part dans le code (le mécanisme actif est le tableau `__linkedProperties`) — potentiellement mort, mais pas supprimé. - `REQUEST_DOM_LINK_KEY` semble ne plus être utilisé nulle part dans le code (le mécanisme actif est le tableau `__linkedProperties`) — potentiellement mort, mais pas supprimé.
- Quelques typos mineures ("atatched", "attriute") et un commentaire en français dans `LinkToDom` qui ne correspond plus exactement au comportement (`throw` au lieu d'un simple avertissement). - Quelques typos mineures ("atatched", "attriute") et un commentaire en français dans `LinkToDom` qui ne correspond plus exactement au comportement (`throw` au lieu d'un simple avertissement).
| `@lightweight-sdk/service-registry` | common | 0.0.1 | brouillon |

3
tsconfig.json
View file

@ -6,6 +6,9 @@
}, },
{ {
"path": "./packages/common/service-registry" "path": "./packages/common/service-registry"
},
{
"path": "./packages/server/service-discovery"
} }
] ]
} }