feat(node-message-broker): resolve analysis participants via server-core

apps/node-message-broker/package.json

@@ -23,6 +23,7 @@
         "node": ">=24"
     },
     "dependencies": {
+        "@privateaim/core-http-kit": "^0.11.5",
         "@privateaim/errors": "^0.8.42",
         "@privateaim/kit": "^0.11.5",
         "@privateaim/messenger-http-kit": "^0.11.5",

apps/node-message-broker/src/adapters/core/index.ts

@@ -0,0 +1,8 @@
+/*
+ * Copyright (c) 2026.
+ * Author Peter Placzek (tada5hi)
+ * For the full copyright and license information,
+ * view the LICENSE file that was distributed with this source code.
+ */
+
+export * from './participant-resolver.ts';

apps/node-message-broker/src/adapters/core/participant-resolver.ts

@@ -0,0 +1,109 @@
+/*
+ * Copyright (c) 2026.
+ * Author Peter Placzek (tada5hi)
+ * For the full copyright and license information,
+ * view the LICENSE file that was distributed with this source code.
+ */
+
+import type { Logger } from '@privateaim/server-kit';
+import { LRUCache } from 'lru-cache';
+import type {
+    AnalysisParticipant,
+    IAnalysisNodeProvider,
+    IParticipantResolver,
+} from '../../core/analysis/index.ts';
+
+const DEFAULT_CACHE_MAX = 256;
+
+const DEFAULT_CACHE_TTL_MS = 30_000;
+
+type ParticipantResolverContext = {
+    provider: IAnalysisNodeProvider,
+    /** this node's own authup client id (config.clientId) — used by {@link resolveSelf} */
+    selfClientId: string,
+    cacheMax?: number,
+    /** per-analysis cache lifetime; `<= 0` disables caching. */
+    cacheTtlMs?: number,
+    logger?: Logger
+};
+
+/**
+ * Resolves analysis participants — their node id / type / client id / public key —
+ * from server-core's analysis-node API, mapping each included `node` to an
+ * {@link AnalysisParticipant}. Results are cached per analysis (short TTL) to keep
+ * resolution off the Hub send/deliver hot path; participants missing a client id
+ * or public key are skipped, since they can't be addressed or encrypted to.
+ */
+export class ParticipantResolver implements IParticipantResolver {
+    protected provider: IAnalysisNodeProvider;
+
+    protected selfClientId: string;
+
+    protected logger: Logger | undefined;
+
+    protected cache: LRUCache<string, Promise<AnalysisParticipant[]>> | undefined;
+
+    constructor(ctx: ParticipantResolverContext) {
+        this.provider = ctx.provider;
+        this.selfClientId = ctx.selfClientId;
+        this.logger = ctx.logger;
+
+        const ttl = ctx.cacheTtlMs ?? DEFAULT_CACHE_TTL_MS;
+        if (ttl > 0) {
+            this.cache = new LRUCache<string, Promise<AnalysisParticipant[]>>({
+                max: Math.max(1, Math.floor(ctx.cacheMax ?? DEFAULT_CACHE_MAX)),
+                ttl,
+            });
+        }
+    }
+
+    resolve(analysisId: string): Promise<AnalysisParticipant[]> {
+        const cached = this.cache?.get(analysisId);
+        if (cached) {
+            return cached;
+        }
+
+        const promise = this.fetch(analysisId);
+
+        const { cache } = this;
+        if (cache) {
+            // evict a rejected fetch so a transient failure isn't cached for the TTL
+            promise.catch(() => {
+                if (cache.peek(analysisId) === promise) {
+                    cache.delete(analysisId);
+                }
+            });
+            cache.set(analysisId, promise);
+        }
+
+        return promise;
+    }
+
+    async resolveSelf(analysisId: string): Promise<AnalysisParticipant | undefined> {
+        const participants = await this.resolve(analysisId);
+        return participants.find((participant) => participant.clientId === this.selfClientId);
+    }
+
+    protected async fetch(analysisId: string): Promise<AnalysisParticipant[]> {
+        const nodes = await this.provider.list(analysisId);
+
+        const participants: AnalysisParticipant[] = [];
+        for (const node of nodes) {
+            if (!node.client_id || !node.public_key) {
+                this.logger?.warn(
+                    `Analysis ${analysisId}: skipping participant ${node.id} with a missing client id or public key`,
+                );
+                continue;
+            }
+
+            participants.push({
+                nodeId: node.id,
+                nodeType: node.type,
+                clientId: node.client_id,
+                publicKey: node.public_key,
+            });
+        }
+
+        return participants;
+    }
+}

apps/node-message-broker/src/app/builder.ts

@@ -9,6 +9,7 @@ import { BaseApplicationBuilder } from '@privateaim/server-kit';
 import type { IModule } from 'orkos';
 import { ConfigModule } from './modules/config/index.ts';
 import { ComponentsModule } from './modules/components/index.ts';
+import { CoreClientModule } from './modules/core-client/index.ts';
 import { HTTPModule } from './modules/http/index.ts';
 
 export class ServerMessageBrokerApplicationBuilder extends BaseApplicationBuilder {
@@ -20,6 +21,10 @@ export class ServerMessageBrokerApplicationBuilder extends BaseApplicationBuilde
         return this.addModuleSlot('components', instance, () => new ComponentsModule());
     }
 
+    withCoreClient(instance?: CoreClientModule | false): this {
+        return this.addModuleSlot('coreClient', instance, () => new CoreClientModule());
+    }
+
     withHTTP(instance?: HTTPModule | false): this {
         return this.addModuleSlot('http', instance, () => new HTTPModule());
     }

apps/node-message-broker/src/app/factory.ts

@@ -18,6 +18,7 @@ export function createApplication(): Application {
             ],
         })
         .withComponents()
+        .withCoreClient()
         .withHTTP();
 
     return builder.build();

apps/node-message-broker/src/app/modules/core-client/constants.ts

@@ -0,0 +1,11 @@
+/*
+ * Copyright (c) 2026.
+ * Author Peter Placzek (tada5hi)
+ * For the full copyright and license information,
+ * view the LICENSE file that was distributed with this source code.
+ */
+
+import { TypedToken } from 'eldin';
+import type { IParticipantResolver } from '../../../core/analysis/index.ts';
+
+export const CoreClientInjectionKey = { ParticipantResolver: new TypedToken<IParticipantResolver>('ParticipantResolver') };

apps/node-message-broker/src/app/modules/core-client/index.ts

@@ -0,0 +1,9 @@
+/*
+ * Copyright (c) 2026.
+ * Author Peter Placzek (tada5hi)
+ * For the full copyright and license information,
+ * view the LICENSE file that was distributed with this source code.
+ */
+
+export * from './constants.ts';
+export * from './module.ts';

apps/node-message-broker/src/app/modules/core-client/module.ts

@@ -0,0 +1,88 @@
+/*
+ * Copyright (c) 2026.
+ * Author Peter Placzek (tada5hi)
+ * For the full copyright and license information,
+ * view the LICENSE file that was distributed with this source code.
+ */
+
+import type { IContainer } from 'eldin';
+import type { IModule } from 'orkos';
+import { Client } from '@privateaim/core-http-kit';
+import {
+    LoggerInjectionKey,
+    createAuthupClientAuthenticationHook,
+    createAuthupClientTokenCreator,
+} from '@privateaim/server-kit';
+import type { IAnalysisNodeProvider } from '../../../core/analysis/index.ts';
+import { ParticipantResolver } from '../../../adapters/core/index.ts';
+import { ConfigInjectionKey } from '../config/constants.ts';
+import { CoreClientInjectionKey } from './constants.ts';
+
+/**
+ * Wires the server-core link: a `@privateaim/core-http-kit` client (authenticated
+ * as the node client via client credentials, mirroring the Hub link) backing the
+ * {@link ParticipantResolver}. Registered for the analysis policy (S3) and the
+ * send/deliver flows (S5/S6) to consume.
+ */
+export class CoreClientModule implements IModule {
+    readonly name = 'coreClient';
+
+    readonly dependencies: string[] = ['config'];
+
+    private authHook: ReturnType<typeof createAuthupClientAuthenticationHook> | undefined;
+
+    async setup(container: IContainer): Promise<void> {
+        const config = container.resolve(ConfigInjectionKey);
+
+        const loggerResult = container.tryResolve(LoggerInjectionKey);
+        const logger = loggerResult.success ? loggerResult.data : undefined;
+
+        const tokenCreator = createAuthupClientTokenCreator({
+            baseURL: config.authupURL,
+            clientId: config.clientId,
+            clientSecret: config.clientSecret,
+            realm: config.realm,
+        });
+
+        const client = new Client({ baseURL: config.coreURL });
+        const authHook = createAuthupClientAuthenticationHook({
+            baseURL: config.authupURL,
+            tokenCreator,
+        });
+        authHook.attach(client);
+        this.authHook = authHook;
+
+        // wrap the core client's analysis-node lookup behind the narrow provider port
+        const provider: IAnalysisNodeProvider = {
+            list: async (analysisId) => {
+                const response = await client.analysisNode.getMany({
+                    filter: { analysis_id: analysisId },
+                    include: ['node'],
+                });
+                // `node` is typed non-null, but server-core may omit the relation
+                // (include not honored / incomplete data); drop those so the
+                // resolver never dereferences an undefined node.
+                return response.data
+                    .map((entry) => entry.node)
+                    .filter((node): node is NonNullable<typeof node> => !!node);
+            },
+        };
+
+        const resolver = new ParticipantResolver({
+            provider,
+            selfClientId: config.clientId,
+            logger,
+        });
+        container.register(CoreClientInjectionKey.ParticipantResolver, { useValue: resolver });
+    }
+
+    async teardown(): Promise<void> {
+        // The auth hook owns a token-refresh timer; drop it so it can't fire (and
+        // hit Authup) after shutdown.
+        if (this.authHook) {
+            this.authHook.disable();
+            this.authHook.clearTimer();
+            this.authHook = undefined;
+        }
+    }
+}

apps/node-message-broker/src/core/analysis/types.ts

@@ -9,9 +9,29 @@
 export type AnalysisParticipant = {
     nodeId: string,
     nodeType: string,
-    clientId: string
+    clientId: string,
+    /** the node's ECDH public key (hex-encoded SPKI PEM) for node-to-node E2E crypto */
+    publicKey: string
 };
 
+/** The slice of a server-core `Node` the broker reads off the analysis-node relation. */
+export type CoreNode = {
+    id: string,
+    type: string,
+    client_id: string | null,
+    public_key: string | null
+};
+
+/**
+ * Lists the nodes participating in an analysis — server-core's analysis-node
+ * relation with the `node` included. Declared as a narrow port so the resolver can
+ * be tested with a fake; the live implementation (in `app/modules/core-client`)
+ * wraps `@privateaim/core-http-kit`'s `analysisNode.getMany`.
+ */
+export interface IAnalysisNodeProvider {
+    list(analysisId: string): Promise<CoreNode[]>;
+}
+
 /**
  * Analysis authorization lives node-side (the Hub is analysis-agnostic). Asserts
  * the calling analysis client holds `ANALYSIS_SELF_MESSAGE_BROKER_USE` — read from

apps/node-message-broker/test/unit/adapters/core/fake-analysis-node-provider.ts

@@ -0,0 +1,30 @@
+/*
+ * Copyright (c) 2026.
+ * Author Peter Placzek (tada5hi)
+ * For the full copyright and license information,
+ * view the LICENSE file that was distributed with this source code.
+ */
+
+import type { CoreNode, IAnalysisNodeProvider } from '../../../../src/core/analysis/index.ts';
+
+/**
+ * In-memory `IAnalysisNodeProvider` that records every lookup and returns
+ * configurable canned nodes per analysis — stands in for the server-core
+ * analysis-node lookup so the resolver is testable without a live core client.
+ */
+export class FakeAnalysisNodeProvider implements IAnalysisNodeProvider {
+    calls: string[] = [];
+
+    nodesByAnalysis = new Map<string, CoreNode[]>();
+
+    /** When set, the next `list` rejects with this error. */
+    error: Error | undefined;
+
+    list = async (analysisId: string): Promise<CoreNode[]> => {
+        this.calls.push(analysisId);
+        if (this.error) {
+            throw this.error;
+        }
+        return this.nodesByAnalysis.get(analysisId) ?? [];
+    };
+}