lee/puter
1
0
Fork 0
mirror of https://github.com/HeyPuter/puter.git synced 2026-08-26 16:07:13 +00:00
Code Issues Actions 9 Packages Projects Releases Wiki Activity

feat: type extension cache update events (#2548)

Browse Source 
* feat: type extension cache update events

Expose outer.cacheUpdate in extension API typings and consolidate extension service typing via ServicesMap.

* fix: batch broadcast events

* fix: bad import

* fix: import socket io

* fix: bad undefined call

* fix: simplify await for broadcast processing
This commit is contained in:
Daniel Salazar
2026-02-25 17:05:05 -08:00
committed by GitHub
parent aa04dfabb4
commit 2c1b21e197
4 changed files with 241 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/api.d.ts
Vendored
+8 -28
View File
@@ -1,33 +1,26 @@
import type APIError from '@heyputer/backend/src/api/APIError.js';
import type { WebServerService } from '@heyputer/backend/src/modules/web/WebServerService.js';
import type query from '@heyputer/backend/src/om/query/query';
import type { Actor } from '@heyputer/backend/src/services/auth/Actor.js';
import type { ServicesMap } from '@heyputer/backend/src/services/BaseService.d.ts';
import type { BaseDatabaseAccessService } from '@heyputer/backend/src/services/database/BaseDatabaseAccessService.d.ts';
import type { EmailService } from '@heyputer/backend/src/services/EmailService.js';
import type { EntityStoreService } from '@heyputer/backend/src/services/EntityStoreService.js';
import type { GetUserService } from '@heyputer/backend/src/services/GetUserService.js';
import type { MeteringService } from '@heyputer/backend/src/services/MeteringService/MeteringService.ts';
import type { MeteringServiceWrapper } from '@heyputer/backend/src/services/MeteringService/MeteringServiceWrapper.mjs';
import type { DynamoKVStore } from '@heyputer/backend/src/services/repositories/DynamoKVStore/DynamoKVStore.ts';
import type { SUService } from '@heyputer/backend/src/services/SUService.js';
import type { IUser } from '@heyputer/backend/src/services/User.js';
import type { UserService } from '@heyputer/backend/src/services/UserService.d.ts';
import type { Context } from '@heyputer/backend/src/util/context.js';
import type kvjs from '@heyputer/kv.js';
import type { RequestHandler } from 'express';
import type { Cluster } from 'ioredis';
import type FSNodeContext from '../src/backend/src/filesystem/FSNodeContext.js';
import type helpers from '../src/backend/src/helpers.js';
import type * as ExtensionControllerExports from './ExtensionController/src/ExtensionController.ts';
import type { ICompleteArguments } from '../src/backend/src/services/ai/chat/providers/types.ts';
import type * as ExtensionControllerExports from './ExtensionController/src/ExtensionController.ts';
declare global {
namespace Express {
interface Request {
services: {
get: <T extends keyof ServiceNameMap | (string & {})>(
get: <T extends keyof ServicesMap | (string & {})>(
string: T,
) => T extends keyof ServiceNameMap ? ServiceNameMap[T] : unknown;
) => T extends keyof ServicesMap ? ServicesMap[T] : unknown;
};
actor?: Actor;
rawBody: Buffer;
@@ -94,20 +87,6 @@ interface FilesystemModule {
selectors: unknown;
}
// TODO DS: define this globally in core to use it there too
interface ServiceNameMap {
meteringService: Pick<MeteringServiceWrapper, 'meteringService'> &
MeteringService; // TODO DS: squash into a single class without wrapper
'puter-kvstore': DynamoKVStore;
su: SUService;
database: BaseDatabaseAccessService;
user: UserService;
'get-user': GetUserService;
'web-server': WebServerService;
'email': EmailService;
'es:app': EntityStoreService;
}
export interface ExtensionEventTypeMap {
'metering:registerAvailablePolicies': {
availablePolicies: unknown[]
@@ -157,6 +136,7 @@ export interface ExtensionEventTypeMap {
intended_service: string,
parameters: ICompleteArguments
}
'outer.cacheUpdate': { cacheKey: string | string[], ttlSeconds?: number, data?: unknown }
}
interface Extension extends RouterMethods {
@@ -182,10 +162,10 @@ interface Extension extends RouterMethods {
import(module: 'fs'): FilesystemModule;
import(module: 'query'): typeof query;
import(module: 'extensionController'): typeof ExtensionControllerExports;
import<T extends `service:${keyof ServiceNameMap}` | (string & {})>(
import<T extends `service:${keyof ServicesMap}` | (string & {})>(
module: T
): T extends `service:${infer R extends keyof ServiceNameMap}`
? ServiceNameMap[R]
): T extends `service:${infer R extends keyof ServicesMap}`
? ServicesMap[R]
: unknown;
}
src/backend/src/modules/broadcast/BroadcastService.js
+187 -118
View File
@@ -16,85 +16,195 @@
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
const BaseService = require('../../services/BaseService');
const { CLink } = require('./connection/CLink');
const { SLink } = require('./connection/SLink');
const { Context } = require('../../util/context');
const { Endpoint } = require('../../util/expressutil');
const crypto = require('crypto');
import { createHmac, timingSafeEqual } from 'crypto';
import { Server as SocketIoServer } from 'socket.io';
import { BaseService } from '../../services/BaseService.js';
import { Context } from '../../util/context.js';
import { Endpoint } from '../../util/expressutil.js';
import { CLink } from './connection/CLink.js';
import { SLink } from './connection/SLink.js';
class BroadcastService extends BaseService {
static MODULES = {
express: require('express'),
// ['socket.io']: require('socket.io'),
};
_construct () {
this.peers_ = [];
this.connections_ = [];
this.trustedPublicKeys_ = {};
this.peersByKey_ = {};
this.webhookPeers_ = [];
this.incomingLastNonceByPeer_ = new Map();
this.outgoingNonceByPeer_ = new Map();
}
export class BroadcastService extends BaseService {
#peers = [];
#connections = [];
#trustedPublicKeys = {};
#peersByKey = {};
#webhookPeers = [];
#incomingLastNonceByPeer = new Map();
#outgoingNonceByPeer = new Map();
#outboundEventsByDedupKey = new Map();
#outboundFlushTimer = null;
#outboundIsFlushing = false;
#dedupFallbackCounter = 0;
#webhookReplayWindowSeconds = 300;
#outboundFlushMs = 5000;
async _init () {
const peers = this.config.peers ?? [];
const replayWindowSeconds = this.config.webhook_replay_window_seconds ?? 300;
const outboundFlushMs = Number(this.config.outbound_flush_ms ?? 5000);
for ( const peer_config of peers ) {
this.trustedPublicKeys_[peer_config.key] = true;
this.peersByKey_[peer_config.key] = {
this.#trustedPublicKeys[peer_config.key] = true;
this.#peersByKey[peer_config.key] = {
webhook_secret: peer_config.webhook_secret,
webhook_url: peer_config.webhook_url,
webhook: !!peer_config.webhook,
};
if ( peer_config.webhook ) {
this.webhookPeers_.push(peer_config);
this.#webhookPeers.push(peer_config);
} else {
const peer = new CLink({
keys: this.config.keys,
config: peer_config,
log: this.log,
});
this.peers_.push(peer);
this.#peers.push(peer);
peer.connect();
}
}
this.webhookReplayWindowSeconds_ = replayWindowSeconds;
this._register_commands(this.services.get('commands'));
this.#webhookReplayWindowSeconds = replayWindowSeconds;
this.#outboundFlushMs = Number.isFinite(outboundFlushMs) && outboundFlushMs >= 0
? outboundFlushMs
: 5000;
const svc_event = this.services.get('event');
svc_event.on('outer.*', this.on_event.bind(this));
// Test event (logs a message to console if DEBUG is set in env)
svc_event.on('test', (key, data, _meta) => {
const { contents } = data;
console.log(`Test Message: ${contents}`);
});
svc_event.on('outer.*', this.outBroadcastEventHandler.bind(this));
}
async on_event (key, data, meta) {
if ( meta.from_outside ) return;
async outBroadcastEventHandler (key, data, meta) {
if ( meta?.from_outside ) return;
for ( const peer of this.peers_ ) {
try {
peer.send({ key, data, meta });
} catch (e) {
//
const safeMeta = this.#normalizeMeta(meta);
this.#enqueueOutboundEvent({ key, data, meta: safeMeta });
}
#enqueueOutboundEvent (event) {
const dedupKey = this.#createDedupKey(event);
this.#outboundEventsByDedupKey.set(dedupKey, event);
this.#scheduleOutboundFlush();
}
#createDedupKey (event) {
try {
return JSON.stringify(event);
} catch {
const fallbackKey = `fallback-${this.#dedupFallbackCounter}`;
this.#dedupFallbackCounter += 1;
return fallbackKey;
}
}
#scheduleOutboundFlush () {
if ( this.#outboundFlushTimer ) return;
this.#outboundFlushTimer = setTimeout(() => {
this.#outboundFlushTimer = null;
this.#flushOutboundEvents().catch(error => {
console.warn('outbound broadcast flush failed', { error });
});
}, this.#outboundFlushMs);
}
async #flushOutboundEvents () {
if ( this.#outboundIsFlushing || this.#outboundEventsByDedupKey.size === 0 ) return;
this.#outboundIsFlushing = true;
try {
const events = [...this.#outboundEventsByDedupKey.values()];
this.#outboundEventsByDedupKey.clear();
const message = { events };
for ( const peer of this.#peers ) {
try {
peer.send(message);
} catch (e) {
console.warn(`ws broadcast send error: ${ JSON.stringify({ peer: peer.key, error: e })}`);
}
}
for ( const peer_config of this.#webhookPeers ) {
try {
await this.#sendWebhookToPeer(peer_config, events);
} catch (e) {
console.warn(`webhook broadcast send error: ${ JSON.stringify({ peer: peer_config.key, error: e })}`);
}
}
} finally {
this.#outboundIsFlushing = false;
if ( this.#outboundEventsByDedupKey.size > 0 ) {
this.#scheduleOutboundFlush();
}
}
}
for ( const peer_config of this.webhookPeers_ ) {
try {
await this.sendWebhookToPeer_(peer_config, key, data, meta);
} catch (e) {
this.log?.warn?.('broadcast webhook send failed', { peer: peer_config.key, error: e });
#normalizeMeta (meta) {
if ( !meta || typeof meta !== 'object' || Array.isArray(meta) ) {
return {};
}
return meta;
}
#normalizeIncomingPayload (payload) {
if ( !payload || typeof payload !== 'object' || Array.isArray(payload) ) {
return null;
}
if ( Array.isArray(payload.events) ) {
const events = [];
for ( const event of payload.events ) {
const normalized = this.#normalizeIncomingEvent(event);
if ( ! normalized ) return null;
events.push(normalized);
}
return events;
}
const normalized = this.#normalizeIncomingEvent(payload);
if ( ! normalized ) return null;
return [normalized];
}
#normalizeIncomingEvent (event) {
if ( !event || typeof event !== 'object' || Array.isArray(event) ) {
return null;
}
const { key, data } = event;
if ( key === undefined || key === null ) {
return null;
}
if ( data === undefined ) {
return null;
}
return {
key,
data,
meta: this.#normalizeMeta(event.meta),
};
}
async #emitIncomingEventsSequentially (events) {
const svcEvent = this.services.get('event');
const context = Context.get(undefined, { allow_fallback: true });
for ( const event of events ) {
if ( event.meta?.from_outside ) {
console.warn('possible over-sending');
continue;
}
if ( event.key === 'test' ) {
console.debug(`test message: ${JSON.stringify(event.data)}`);
}
const metaOut = { ...event.meta, from_outside: true };
await context.arun(async () => {
await svcEvent.emit(event.key, event.data, metaOut);
});
}
}
@@ -105,11 +215,11 @@ class BroadcastService extends BaseService {
Endpoint({
route: '/broadcast/webhook',
methods: ['POST'],
handler: this.handleWebhookRequest_.bind(this),
handler: this.#handleWebhookRequest.bind(this),
}).attach(app);
}
async handleWebhookRequest_ (req, res) {
async #handleWebhookRequest (req, res) {
const rawBody = req.rawBody;
if ( rawBody === undefined || rawBody === null ) {
res.status(400).send({ error: { message: 'Missing or invalid body' } });
@@ -122,18 +232,9 @@ class BroadcastService extends BaseService {
return;
}
// Validate required properties
const { key, data, meta } = body;
if ( key === undefined || key === null ) {
res.status(400).send({ error: { message: 'Missing key' } });
return;
}
if ( data === undefined ) {
res.status(400).send({ error: { message: 'Missing data' } });
return;
}
if ( meta === undefined ) {
res.status(400).send({ error: { message: 'Missing meta' } });
const incomingEvents = this.#normalizeIncomingPayload(body);
if ( ! incomingEvents ) {
res.status(400).send({ error: { message: 'Invalid broadcast payload' } });
return;
}
@@ -143,9 +244,9 @@ class BroadcastService extends BaseService {
return;
}
this.log.debug('received peerId', { value: peerId });
console.debug('received peerId', { value: peerId });
const peer = this.peersByKey_[peerId];
const peer = this.#peersByKey[peerId];
if ( !peer || !peer.webhook_secret ) {
res.status(403).send({ error: { message: 'Unknown peer or webhook not configured' } });
return;
@@ -163,7 +264,7 @@ class BroadcastService extends BaseService {
return;
}
const nowSeconds = Math.floor(Date.now() / 1000);
const window = this.webhookReplayWindowSeconds_;
const window = this.#webhookReplayWindowSeconds;
if ( timestamp < nowSeconds - window || timestamp > nowSeconds + 60 ) {
res.status(400).send({ error: { message: 'Timestamp out of window' } });
return;
@@ -180,7 +281,7 @@ class BroadcastService extends BaseService {
res.status(400).send({ error: { message: 'Invalid X-Broadcast-Nonce' } });
return;
}
const lastNonce = this.incomingLastNonceByPeer_.get(peerId) ?? -1;
const lastNonce = this.#incomingLastNonceByPeer.get(peerId) ?? -1;
if ( nonce <= lastNonce ) {
res.status(403).send({ error: { message: 'Duplicate or stale nonce' } });
return;
@@ -194,48 +295,38 @@ class BroadcastService extends BaseService {
}
const payloadToSign = `${timestamp}.${nonce}.${rawBody}`;
const expectedHmac = crypto.createHmac('sha256', peer.webhook_secret).update(payloadToSign).digest('hex');
const expectedHmac = createHmac('sha256', peer.webhook_secret).update(payloadToSign).digest('hex');
const signatureBuffer = Buffer.from(signatureHeader, 'hex');
const expectedBuffer = Buffer.from(expectedHmac, 'hex');
if ( signatureBuffer.length !== expectedBuffer.length || !crypto.timingSafeEqual(signatureBuffer, expectedBuffer) ) {
if ( signatureBuffer.length !== expectedBuffer.length || !timingSafeEqual(signatureBuffer, expectedBuffer) ) {
res.status(403).send({ error: { message: 'Invalid signature' } });
return;
}
this.incomingLastNonceByPeer_.set(peerId, nonce);
this.#incomingLastNonceByPeer.set(peerId, nonce);
// We emit the event sent to this webhook endpoint so other services
// can react to it. We set the `from_outside` flag to avoid feedback.
const svc_event = this.services.get('event');
const metaOut = { ...meta, from_outside: true };
const context = Context.get(undefined, { allow_fallback: true });
await context.arun(async () => {
this.log.debug('Emitting to the event service', {
key, data, metaOut,
});
await svc_event.emit(key, data, metaOut);
});
this.#emitIncomingEventsSequentially(incomingEvents);
res.status(200).send({ ok: true });
}
async sendWebhookToPeer_ (peer_config, key, data, meta) {
async #sendWebhookToPeer (peer_config, events) {
const peerId = peer_config.key;
const url = peer_config.webhook_url;
const mySecretKey = this.config.webhook?.secret ?? '';
if ( !url || !mySecretKey ) return;
let nextNonce = this.outgoingNonceByPeer_.get(peerId) ?? 0;
this.outgoingNonceByPeer_.set(peerId, nextNonce + 1);
let nextNonce = this.#outgoingNonceByPeer.get(peerId) ?? 0;
this.#outgoingNonceByPeer.set(peerId, nextNonce + 1);
const timestamp = Math.floor(Date.now() / 1000);
const body = { key, data, meta };
const body = { events };
const rawBody = JSON.stringify(body);
const payloadToSign = `${timestamp}.${nextNonce}.${rawBody}`;
const signature = crypto.createHmac('sha256', mySecretKey).update(payloadToSign).digest('hex');
const signature = createHmac('sha256', mySecretKey).update(payloadToSign).digest('hex');
const myPublicKey = this.config.webhook?.key ?? '';
this.log.debug('Sending webhook message to peer', { peerId });
console.debug('Sending webhook message to peer', { peerId });
const response = await fetch(url, {
method: 'POST',
headers: {
@@ -254,12 +345,11 @@ class BroadcastService extends BaseService {
}
async '__on_install.websockets' () {
const svc_event = this.services.get('event');
const svc_webServer = this.services.get('web-server');
const server = svc_webServer.get_server();
const io = require('socket.io')(server, {
const io = new SocketIoServer(server, {
cors: { origin: '*' },
path: '/wssinternal',
});
@@ -267,45 +357,24 @@ class BroadcastService extends BaseService {
io.on('connection', async socket => {
const conn = new SLink({
keys: this.config.keys,
trustedKeys: this.trustedPublicKeys_,
trustedKeys: this.#trustedPublicKeys,
socket,
});
this.connections_.push(conn);
this.#connections.push(conn);
conn.channels.message.on(({ key, data, meta }) => {
if ( meta.from_outside ) {
console.warn('possible over-sending');
conn.channels.message.on(async message => {
const incomingEvents = this.#normalizeIncomingPayload(message);
if ( ! incomingEvents ) {
console.warn('invalid ws broadcast payload');
return;
}
if ( key === 'test' ) {
console.debug(`test message: ${
JSON.stringify(data)}`);
try {
await this.#emitIncomingEventsSequentially(incomingEvents);
} catch ( error ) {
console.warn('ws broadcast receive error', { error });
}
meta.from_outside = true;
const context = Context.get(undefined, { allow_fallback: true });
context.arun(async () => {
await svc_event.emit(key, data, meta);
});
});
});
}
_register_commands (commands) {
commands.registerCommands('broadcast', [
{
id: 'test',
description: 'send a test message',
handler: async () => {
this.log.info('broadcast service test command was run');
this.on_event('test', {
contents: 'I am a test message',
}, {});
},
},
]);
}
}
module.exports = { BroadcastService };
src/backend/src/services/BaseService.d.ts
Vendored
+45 -26
View File
@@ -1,35 +1,54 @@
import type { ErrorService } from '@heyputer/backend/src/modules/core/ErrorService';
import { DDBClient } from '../clients/dynamodb/DDBClient';
import type { DriverService } from '@heyputer/backend/src/services/drivers/DriverService';
import type { DynamoKVStore } from '@heyputer/backend/src/services/DynamoKVStore/DynamoKVStore';
import type { DDBClient } from '../clients/dynamodb/DDBClient';
import type { ServerHealthService } from '../modules/core/ServerHealthService/ServerHealthService';
import { GroupService } from './auth/GroupService';
import { SignupService} from './auth/SignupService';
import { CleanEmailService } from './CleanEmailService';
import { SqliteDatabaseAccessService } from './database/SqliteDatabaseAccessService';
import { EventService } from './EventService';
import { FeatureFlagService } from './FeatureFlagService';
import { MeteringServiceWrapper } from './MeteringService/MeteringServiceWrapper.mjs';
import type { WebServerService } from '../modules/web/WebServerService';
import type { GroupService } from './auth/GroupService';
import type { SignupService } from './auth/SignupService';
import type { CleanEmailService } from './CleanEmailService';
import type { SqliteDatabaseAccessService } from './database/SqliteDatabaseAccessService';
import type { IDynamoKVStoreWrapper } from './DynamoKVStore/DynamoKVStoreWrapper';
import type { Emailservice } from './EmailService';
import type { EntityStoreService } from './EntityStoreService';
import type { EventService } from './EventService';
import type { FeatureFlagService } from './FeatureFlagService';
import type { GetUserService } from './GetUserService';
import type { MeteringService } from './MeteringService/MeteringService';
import type { MeteringServiceWrapper } from './MeteringService/MeteringServiceWrapper.mjs';
import type { SUService } from './SUService';
import { UserService } from './UserService';
import type{ DynamoKVStore } from '@heyputer/backend/src/services/DynamoKVStore/DynamoKVStore';
import { DriverService } from '@heyputer/backend/src/services/drivers/DriverService';
import type { UserService } from './UserService';
export interface ServicesMap {
su: SUService;
user: UserService;
'get-user': GetUserService;
'web-server': WebServerService;
email: Emailservice;
'es:app': EntityStoreService;
meteringService: MeteringService & MeteringServiceWrapper;
'puter-kvstore': DynamoKVStore & IDynamoKVStoreWrapper;
database: SqliteDatabaseAccessService;
'server-health': ServerHealthService;
su: SUService;
dynamo: DDBClient;
user: UserService;
event: EventService;
signup: SignupService;
group: GroupService;
'feature-flag': FeatureFlagService;
'clean-email': CleanEmailService;
'error-service': ErrorService;
driver: DriverService;
}
export interface ServiceResources {
services: {
get (name: 'meteringService'): MeteringServiceWrapper;
get (name: 'puter-kvstore'): DynamoKVStore;
get (name: 'database'): SqliteDatabaseAccessService;
get (name: 'server-health'): ServerHealthService;
get (name: 'su'): SUService;
get (name: 'dynamo'): DDBClient;
get (name: 'user'): UserService;
get (name: 'event'): EventService;
get (name: 'signup'): SignupService;
get (name: 'group'): GroupService;
get (name: 'feature-flag'): FeatureFlagService;
get (name: 'clean-email'): CleanEmailService;
get (name: 'error-service'): ErrorService;
get (name: 'driver'): DriverService;
get (name: string): unknown;
get<T extends `${keyof ServicesMap}` | (string & {})>(
name: T
): T extends `${infer R extends keyof ServicesMap}`
? ServicesMap[R]
: unknown;
};
config: Record<string, any> & { services?: Record<string, any>; server_id?: string };
name?: string;
src/backend/src/services/DynamoKVStore/DynamoKVStoreWrapper.ts
+1 -1
View File
@@ -45,7 +45,7 @@ class DynamoKVStoreServiceWrapper extends BaseService {
}
static IMPLEMENTS = {
['puter-kvstore']: Object.getOwnPropertyNames(DynamoKVStore.prototype)
'puter-kvstore': Object.getOwnPropertyNames(DynamoKVStore.prototype)
.filter(n => n !== 'constructor')
.reduce((acc, fn) => ({
...acc,