import StreamID from '@ceramicnetwork/streamid' ; import CID from 'cids' ; import { UnreachableCaseError } from '@ceramicnetwork/common' ; import dagCBOR from 'ipld-dag-cbor' ; import * as multihashes from 'multihashes' ; import * as sha256 from '@stablelib/sha256' ; import { TextDecoder , TextEncoder } from 'util' ; import * as uint8arrays from 'uint8arrays' ; /**
 * Ceramic Pub/Sub message type.
 */ export enum MsgType { UPDATE , QUERY , RESPONSE , KEEPALIVE } export type UpdateMessage = { typ : MsgType . UPDATE ; stream : StreamID ; tip : CID ; } ; export type QueryMessage = { typ : MsgType . QUERY ; id : string ; stream : StreamID ; } ; export type ResponseMessage = { typ : MsgType . RESPONSE ; id : string ; tips : Map < string , CID > ; } ; export type KeepaliveMessage = { typ : MsgType . KEEPALIVE ; ts : number ; } ; export type PubsubMessage = UpdateMessage | QueryMessage | ResponseMessage | KeepaliveMessage ; const textEncoder = new TextEncoder ( ) ; const textDecoder = new TextDecoder ( 'utf-8' ) ; function messageHash ( message ) { const encoded = dagCBOR . util . serialize ( message ) ; const id = sha256 . hash ( encoded ) ; return uint8arrays . toString ( multihashes . encode ( id , 'sha2-256' ) , 'base64url' ) ; } export function buildQueryMessage ( streamId ) { const payload = { typ : MsgType . QUERY as MsgType . QUERY , stream : streamId , } ; const id = messageHash ( { ... payload , stream : streamId . toString ( ) } ) ; return { ... payload , id : id , } ; } export function serialize ( message ) { switch ( message . typ ) { case MsgType . QUERY : { return textEncoder . encode ( JSON . stringify ( { ... message , doc : message . stream . toString ( ) , stream : message . stream . toString ( ) , } ) ) ; } case MsgType . RESPONSE : { const tips = { } ; message . tips . forEach ( ( value , key ) => ( tips [ key ] = value . toString ( ) ) ) ; const payload = { ... message , tips : tips , } ; return textEncoder . encode ( JSON . stringify ( payload ) ) ; } case MsgType . UPDATE : { const payload = { typ : MsgType . UPDATE , doc : message . stream . toString ( ) , stream : message . stream . toString ( ) , tip : message . tip . toString ( ) , } ; return textEncoder . encode ( JSON . stringify ( payload ) ) ; } case MsgType . KEEPALIVE : { const payload = { typ : MsgType . KEEPALIVE , ts : message . ts , } ; return textEncoder . encode ( JSON . stringify ( payload ) ) ; } default : throw new UnreachableCaseError ( message , 'Unknown message type' ) ; } } export function deserialize ( message ) { const asString = textDecoder . decode ( message . data ) ; const parsed = JSON . parse ( asString ) ; const typ = parsed . typ as MsgType ; switch ( typ ) { case MsgType . UPDATE : { const stream = StreamID . fromString ( parsed . stream || parsed . doc ) ; return { typ : MsgType . UPDATE , stream , tip : new CID ( parsed . tip ) , } ; } case MsgType . RESPONSE : { const tips = new Map ( ) ; Object . entries < string > ( parsed . tips ) . forEach ( ( [ key , value ] ) => tips . set ( key , new CID ( value ) ) ) ; return { typ : MsgType . RESPONSE , id : parsed . id , tips : tips , } ; } case MsgType . QUERY : { const stream = StreamID . fromString ( parsed . stream || parsed . doc ) ; return { typ : MsgType . QUERY , id : parsed . id , stream , } ; } case MsgType . KEEPALIVE : { return { typ : MsgType . KEEPALIVE , ts : parsed . ts , } ; } default : throw new UnreachableCaseError ( typ , 'Unknown message type' ) ; } }