From 3cb31aede51c7fd6a520bdad15d2f78db766c335 Mon Sep 17 00:00:00 2001 From: nugaon Date: Wed, 3 Jun 2026 12:26:16 +0200 Subject: [PATCH 1/2] feat: add MIC and MOC messaging support Implement Mined ID Chunk (MIC) and Mined Owner Chunk (MOC) messaging functionality with mining, sending, and subscription capabilities for both message types. --- src/bee.ts | 270 +++++++++++++++++++++++++++++++++++++++++++++ src/modules/mic.ts | 20 ++++ src/modules/moc.ts | 20 ++++ src/types/index.ts | 22 ++++ src/utils/type.ts | 22 ++++ 5 files changed, 354 insertions(+) create mode 100644 src/modules/mic.ts create mode 100644 src/modules/moc.ts diff --git a/src/bee.ts b/src/bee.ts index 9ffb3729..0043d823 100644 --- a/src/bee.ts +++ b/src/bee.ts @@ -27,6 +27,8 @@ import { postEnvelope } from './modules/envelope' import { FeedPayloadResult, createFeedManifest, fetchLatestFeedUpdate } from './modules/feed' import * as grantee from './modules/grantee' import * as gsoc from './modules/gsoc' +import * as mic from './modules/mic' +import * as moc from './modules/moc' import * as pinning from './modules/pinning' import * as pss from './modules/pss' import { rchash } from './modules/rchash' @@ -59,6 +61,10 @@ import type { LastCashoutActionResponse, LastChequesForPeerResponse, LastChequesResponse, + MicMessageHandler, + MicSubscription, + MocMessageHandler, + MocSubscription, NodeAddresses, NodeInfo, NumberString, @@ -119,6 +125,8 @@ import { prepareDownloadOptions, prepareFileUploadOptions, prepareGsocMessageHandler, + prepareMicMessageHandler, + prepareMocMessageHandler, preparePostageBatchOptions, preparePssMessageHandler, prepareRedundantUploadOptions, @@ -1287,6 +1295,268 @@ export class Bee { return subscription } + /** + * Mines the signer (a private key) to be used to send MOC messages to the specific target overlay address. + * + * Use {@link mocSend} to send MOC messages with the mined signer. + * + * Use {@link mocSubscribe} to subscribe to MOC messages for the identifier. + * + * **Warning! Only full nodes can accept MOC messages.** + * + * @param targetOverlay + * @param identifier + * @param proximity + * @returns + * + * @example + * const identifier = NULL_IDENTIFIER + * const { overlay } = await bee.getNodeAddresses() + * const signer = bee.mocMine(overlay, identifier) + * const cac = makeContentAddressedChunk('Hello MOC!') + * const soc = cac.toSingleOwnerChunk(identifier, signer) + * await bee.mocSend(soc, postageBatchId) + */ + mocMine( + targetOverlay: PeerAddress | Uint8Array | string, + identifier: Identifier | Uint8Array | string, + proximity = 12, + ): PrivateKey { + targetOverlay = new PeerAddress(targetOverlay) + identifier = new Identifier(identifier) + for (let i = 0; i < 0xffff; i++) { + const randomBytes = crypto.getRandomValues(new Uint8Array(32)) + const signer = new PrivateKey(randomBytes) + const socAddress = makeSOCAddress(identifier, signer.publicKey().address()) + const actualProximity = Binary.proximity(socAddress.toUint8Array(), targetOverlay.toUint8Array()) + + if (actualProximity >= proximity) { + return signer + } + } + throw Error('Could not mine a valid signer') + } + + mocSend = moc.send + + /** + * Subscribes to MOC (Mined Owner Chunk) messages for the specified identifier. + * + * The node delivers every incoming single-owner chunk whose identifier matches, + * regardless of its owner. + * + * Use {@link mocSend} to send MOC messages on the identifier. + * + * **Warning! Only full nodes can accept MOC messages.** + * + * @param identifier + * @param handler + * @returns + * + * @example + * const identifier = NULL_IDENTIFIER + * const subscription = bee.mocSubscribe(identifier, { + * onMessage(message) { + * // handle + * }, + * onError(error) { + * // handle + * }, + * onClose() { + * // handle + * } + * }) + */ + mocSubscribe(identifier: Identifier | Uint8Array | string, handler: MocMessageHandler): MocSubscription { + identifier = new Identifier(identifier) + handler = prepareMocMessageHandler(handler) + + const ws = moc.subscribe(this.url, identifier, this.requestOptions.headers) + + let cancelled = false + const cancel = () => { + if (!cancelled) { + cancelled = true + + if (ws.terminate) { + ws.terminate() + } else { + ws.close() + } + } + } + + const subscription = { + identifier, + cancel, + } + + ws.onmessage = async event => { + const data = await prepareWebsocketData(event.data) + + if (data.length) { + handler.onMessage(new Bytes(data), subscription) + } + } + ws.onerror = event => { + if (!cancelled) { + handler.onError(new BeeError(event.message), subscription) + } + } + ws.onclose = () => { + handler.onClose(subscription) + } + + return subscription + } + + /** + * Mines an identifier to be used to send MIC messages to the specified target overlay address. + * + * Unlike {@link gsocMine} (which mines the owner for a fixed identifier), this mines the + * identifier for a fixed owner (the signer), so that the resulting single-owner chunk address + * lands in the target overlay's neighbourhood. This is the MIC (Mined ID Chunk) counterpart. + * + * Use {@link micSend} to send MIC messages with the mined identifier. + * + * Use {@link micSubscribe} to subscribe to MIC messages for the owner (of the signer). + * + * **Warning! Only full nodes can accept MIC messages.** + * + * @param targetOverlay + * @param signer The fixed publisher identity + * @param proximity + * @returns + * + * @example + * const signer = new PrivateKey('...') + * const { overlay } = await bee.getNodeAddresses() + * const identifier = bee.micMine(overlay, signer) + * const cac = makeContentAddressedChunk('MIC!') + * const soc = cac.toSingleOwnerChunk(identifier, signer) + * await bee.micSend(soc, postageBatchId) + */ + micMine( + targetOverlay: PeerAddress | Uint8Array | string, + signer: PrivateKey | Uint8Array | string, + proximity = 12, + ): Identifier { + targetOverlay = new PeerAddress(targetOverlay) + signer = new PrivateKey(signer) + const owner = signer.publicKey().address() + for (let i = 0; i < 0xffff; i++) { + const identifier = new Identifier(crypto.getRandomValues(new Uint8Array(32))) + const socAddress = makeSOCAddress(identifier, owner) + const actualProximity = Binary.proximity(socAddress.toUint8Array(), targetOverlay.toUint8Array()) + + if (actualProximity >= proximity) { + return identifier + } + } + throw Error('Could not mine a valid identifier') + } + + /** + * Sends a MIC (Mined ID Chunk) message with the specified signer and identifier. + * + * A MIC is matched on the receiving node by its owner, regardless of identifier. The + * owner is a fixed publisher identity (the signer); the identifier is mined so that the + * chunk address lands in the subscriber's neighbourhood. Use {@link micMine} to mine + * such an identifier for the subscriber's overlay address. + * + * Use {@link micSubscribe} to subscribe to MIC messages for the owner. + * + * **Warning! Only full nodes can accept MIC messages.** + * + * @param postageBatchId + * @param signer The fixed publisher identity + * @param identifier The mined identifier, typically from {@link micMine} + * @param data + * @param options + * @param requestOptions Options for making requests, such as timeouts, custom HTTP agents, headers, etc. + * @returns + * + * @example + * const signer = new PrivateKey('...') + * const { overlay } = await bee.getNodeAddresses() + * const identifier = bee.micMine(overlay, signer) + * const cac = makeContentAddressedChunk('MIC!') + * const soc = cac.toSingleOwnerChunk(identifier, signer) + * await bee.micSend(soc, postageBatchId) + */ + micSend = mic.send + + /** + * Subscribes to MIC (Mined ID Chunk) messages for the specified owner ethereum address. + * + * The node delivers every incoming single-owner chunk whose owner matches, + * regardless of its identifier. + * + * Use {@link micSend} to send MIC messages for the owner. + * + * **Warning! Only full nodes can accept MIC messages.** + * + * @param address Owner ethereum address (of the publisher's signer) + * @param handler + * @returns + * + * @example + * const signer = new PrivateKey('...') + * const subscription = bee.micSubscribe(signer.publicKey().address(), { + * onMessage(message) { + * // handle + * }, + * onError(error) { + * // handle + * }, + * onClose() { + * // handle + * } + * }) + */ + micSubscribe(address: EthAddress | Uint8Array | string, handler: MicMessageHandler): MicSubscription { + address = new EthAddress(address) + handler = prepareMicMessageHandler(handler) + + const ws = mic.subscribe(this.url, address, this.requestOptions.headers) + + let cancelled = false + const cancel = () => { + if (!cancelled) { + cancelled = true + + if (ws.terminate) { + ws.terminate() + } else { + ws.close() + } + } + } + + const subscription = { + owner: address, + cancel, + } + + ws.onmessage = async event => { + const data = await prepareWebsocketData(event.data) + + if (data.length) { + handler.onMessage(new Bytes(data), subscription) + } + } + ws.onerror = event => { + if (!cancelled) { + handler.onError(new BeeError(event.message), subscription) + } + } + ws.onclose = () => { + handler.onClose(subscription) + } + + return subscription + } + /** * Creates a feed manifest chunk and returns the reference to it. * diff --git a/src/modules/mic.ts b/src/modules/mic.ts new file mode 100644 index 00000000..099aee0a --- /dev/null +++ b/src/modules/mic.ts @@ -0,0 +1,20 @@ +import { System } from 'cafe-utility' +import WebSocket from 'isomorphic-ws' +import { uploadSingleOwnerChunk } from '../chunk/soc' +import { EthAddress } from '../utils/typed-bytes' + +const endpoint = 'mic' + +export { uploadSingleOwnerChunk as send } + +export function subscribe(url: string, owner: EthAddress, headers?: Record) { + const wsUrl = url.replace(/^http/i, 'ws') + + if (System.whereAmI() === 'browser') { + return new WebSocket(`${wsUrl}/${endpoint}/subscribe/${owner.toHex()}`) + } + + return new WebSocket(`${wsUrl}/${endpoint}/subscribe/${owner.toHex()}`, { + headers, + }) +} diff --git a/src/modules/moc.ts b/src/modules/moc.ts new file mode 100644 index 00000000..a17134f8 --- /dev/null +++ b/src/modules/moc.ts @@ -0,0 +1,20 @@ +import { System } from 'cafe-utility' +import WebSocket from 'isomorphic-ws' +import { uploadSingleOwnerChunk } from '../chunk/soc' +import { Identifier } from '../utils/typed-bytes' + +const endpoint = 'moc' + +export { uploadSingleOwnerChunk as send } + +export function subscribe(url: string, identifier: Identifier, headers?: Record) { + const wsUrl = url.replace(/^http/i, 'ws') + + if (System.whereAmI() === 'browser') { + return new WebSocket(`${wsUrl}/${endpoint}/subscribe/${identifier.toHex()}`) + } + + return new WebSocket(`${wsUrl}/${endpoint}/subscribe/${identifier.toHex()}`, { + headers, + }) +} diff --git a/src/types/index.ts b/src/types/index.ts index b6fd5835..ac4d265d 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -381,6 +381,28 @@ export interface GsocMessageHandler { onClose: (subscription: GsocSubscription) => void } +export interface MocSubscription { + readonly identifier: Identifier + cancel: () => void +} + +export interface MocMessageHandler { + onMessage: (message: Bytes, subscription: MocSubscription) => void + onError: (error: BeeError, subscription: MocSubscription) => void + onClose: (subscription: MocSubscription) => void +} + +export interface MicSubscription { + readonly owner: EthAddress + cancel: () => void +} + +export interface MicMessageHandler { + onMessage: (message: Bytes, subscription: MicSubscription) => void + onError: (error: BeeError, subscription: MicSubscription) => void + onClose: (subscription: MicSubscription) => void +} + export interface ReferenceResponse { reference: Reference } diff --git a/src/utils/type.ts b/src/utils/type.ts index 46ce6bba..50a977c1 100644 --- a/src/utils/type.ts +++ b/src/utils/type.ts @@ -7,6 +7,8 @@ import { DownloadOptions, FileUploadOptions, GsocMessageHandler, + MicMessageHandler, + MocMessageHandler, NumberString, PostageBatchOptions, PssMessageHandler, @@ -155,6 +157,26 @@ export function prepareGsocMessageHandler(value: unknown): GsocMessageHandler { } } +export function prepareMocMessageHandler(value: unknown): MocMessageHandler { + const object = Types.asObject(value, { name: 'MocMessageHandler' }) + + return { + onMessage: Types.asFunction(object.onMessage, { name: 'onMessage' }) as MocMessageHandler['onMessage'], + onError: Types.asFunction(object.onError, { name: 'onError' }) as MocMessageHandler['onError'], + onClose: Types.asFunction(object.onClose, { name: 'onClose' }) as MocMessageHandler['onClose'], + } +} + +export function prepareMicMessageHandler(value: unknown): MicMessageHandler { + const object = Types.asObject(value, { name: 'MicMessageHandler' }) + + return { + onMessage: Types.asFunction(object.onMessage, { name: 'onMessage' }) as MicMessageHandler['onMessage'], + onError: Types.asFunction(object.onError, { name: 'onError' }) as MicMessageHandler['onError'], + onClose: Types.asFunction(object.onClose, { name: 'onClose' }) as MicMessageHandler['onClose'], + } +} + export function preparePostageBatchOptions(value: unknown): PostageBatchOptions { const object = Types.asObject(value, { name: 'PostageBatchOptions' }) From 4b4fa57fce4611ce340ead02d3d8fbb27d06084e Mon Sep 17 00:00:00 2001 From: nugaon Date: Wed, 3 Jun 2026 15:13:22 +0200 Subject: [PATCH 2/2] test: micmoc integration --- test/integration/mic.spec.ts | 33 +++++++++++++++++++++++++++++++++ test/integration/moc.spec.ts | 31 +++++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+) create mode 100644 test/integration/mic.spec.ts create mode 100644 test/integration/moc.spec.ts diff --git a/test/integration/mic.spec.ts b/test/integration/mic.spec.ts new file mode 100644 index 00000000..a1a80b05 --- /dev/null +++ b/test/integration/mic.spec.ts @@ -0,0 +1,33 @@ +import { Strings } from 'cafe-utility' +import { Bytes, MicSubscription } from '../../src' +import { PrivateKey } from '../../src/utils/typed-bytes' +import { batch, makeBee } from '../utils' + +test('MIC - end to end test', async () => { + const bee = makeBee() + const { overlay } = await bee.getNodeAddresses() + + // The owner is a fixed publisher identity; the identifier is mined so that the + // chunk address lands in the subscriber's neighbourhood. + const signer = new PrivateKey(Strings.randomHex(64)) + const identifier = bee.micMine(overlay, signer) + + const promise = new Promise<{ subscription: MicSubscription; message: Bytes }>((resolve, reject) => { + const subscription = bee.micSubscribe(signer.publicKey().address(), { + onMessage(message) { + resolve({ subscription, message }) + }, + onError(error) { + reject(error) + }, + onClose() { + void 0 + }, + }) + }) + + await bee.makeSOCWriter(signer).upload(batch(), identifier, new TextEncoder().encode('MIC!')) + const { subscription, message } = await promise + expect(message.toUtf8()).toBe('MIC!') + subscription.cancel() +}) diff --git a/test/integration/moc.spec.ts b/test/integration/moc.spec.ts new file mode 100644 index 00000000..8c0465a7 --- /dev/null +++ b/test/integration/moc.spec.ts @@ -0,0 +1,31 @@ +import { Bytes, MocSubscription, NULL_IDENTIFIER } from '../../src' +import { batch, makeBee } from '../utils' + +test('MOC - end to end test', async () => { + const bee = makeBee() + const identifier = NULL_IDENTIFIER + const { overlay } = await bee.getNodeAddresses() + + // The identifier is the shared topic; the owner is an ephemeral key mined so that + // the chunk address lands in the subscriber's neighbourhood. + const signer = bee.gsocMine(overlay, identifier) + + const promise = new Promise<{ subscription: MocSubscription; message: Bytes }>((resolve, reject) => { + const subscription = bee.mocSubscribe(identifier, { + onMessage(message) { + resolve({ subscription, message }) + }, + onError(error) { + reject(error) + }, + onClose() { + void 0 + }, + }) + }) + + await bee.makeSOCWriter(signer).upload(batch(), identifier, new TextEncoder().encode('MOC!')) + const { subscription, message } = await promise + expect(message.toUtf8()).toBe('MOC!') + subscription.cancel() +})