diff --git a/extensions/api.d.ts b/extensions/api.d.ts index 63cc45896..add041098 100644 --- a/extensions/api.d.ts +++ b/extensions/api.d.ts @@ -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: ( + get: ( 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 & - 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( + import( module: T - ): T extends `service:${infer R extends keyof ServiceNameMap}` - ? ServiceNameMap[R] + ): T extends `service:${infer R extends keyof ServicesMap}` + ? ServicesMap[R] : unknown; } diff --git a/src/backend/src/modules/broadcast/BroadcastService.js b/src/backend/src/modules/broadcast/BroadcastService.js index fb11cfb47..1c0c9b1bf 100644 --- a/src/backend/src/modules/broadcast/BroadcastService.js +++ b/src/backend/src/modules/broadcast/BroadcastService.js @@ -16,85 +16,195 @@ * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ -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 }; diff --git a/src/backend/src/services/BaseService.d.ts b/src/backend/src/services/BaseService.d.ts index 619af53d6..a1cb8afe4 100644 --- a/src/backend/src/services/BaseService.d.ts +++ b/src/backend/src/services/BaseService.d.ts @@ -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( + name: T + ): T extends `${infer R extends keyof ServicesMap}` + ? ServicesMap[R] + : unknown; }; config: Record & { services?: Record; server_id?: string }; name?: string; diff --git a/src/backend/src/services/DynamoKVStore/DynamoKVStoreWrapper.ts b/src/backend/src/services/DynamoKVStore/DynamoKVStoreWrapper.ts index 4c5e22259..d9dad616c 100644 --- a/src/backend/src/services/DynamoKVStore/DynamoKVStoreWrapper.ts +++ b/src/backend/src/services/DynamoKVStore/DynamoKVStoreWrapper.ts @@ -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,