Add messaging protocol

__tests__/core.ts

@@ -6,6 +6,8 @@ import {
   // protocols
   readContact,
   writeContact,
+  createMailboxPublisher,
+  createMailboxReader,
   FileSystemReader,
   FileSystemWriter,
   downloadFile,
@@ -233,6 +235,51 @@ describe('protocols', () => {
     expect(receivedContact).toEqual(sendContact)
   })
 
+  test('messaging', async done => {
+    jest.setTimeout(10000)
+
+    const aliceMailboxKeyPair = createKeyPair()
+    const bobKeyPair = createKeyPair()
+
+    const publish = createMailboxPublisher({
+      bzz,
+      keyPair: aliceMailboxKeyPair,
+      reader: bobKeyPair.getPublic('hex'),
+    })
+
+    const firstMessage = { title: 'test', body: 'first' }
+    const chapter = await publish(firstMessage)
+
+    const reader = createMailboxReader({
+      bzz,
+      keyPair: bobKeyPair,
+      writer: aliceMailboxKeyPair.getPublic('hex'),
+    })
+
+    const firstChapter = await reader.getLatestChapter()
+    expect(firstChapter).toBeDefined()
+    expect(firstChapter.content.data).toEqual(firstMessage)
+
+    const secondMessage = { thread: chapter.id, title: 'test', body: 'second' }
+
+    const sub = reader.pollLatestChapter({ interval: 1000 }).subscribe({
+      next: chapter => {
+        const { data } = chapter.content
+        if (data.thread != null) {
+          expect(data).toEqual(secondMessage)
+          sub.unsubscribe()
+          done()
+        }
+      },
+      error: err => {
+        sub.unsubscribe()
+        throw err
+      },
+    })
+
+    await publish(secondMessage)
+  })
+
   describe('fileSystem', () => {
     test('uploadFile() and downloadFile()', async () => {
       const data = 'Hello test'

packages/core/src/channels.ts

@@ -240,33 +240,37 @@ export function createEntityFeedSubscriber<T>(params: SubscriberParams) {
 
 export function createEntityTimelineDecoder(params?: DecodeParams) {
   return async function decode(res: Readable) {
-    const stream = await getBodyStream(res, params)
+    const stream = await getBodyStream(res.body, params)
     const body = await getStream(stream)
     const chapter = validateChapter(JSON.parse(body))
     await validateEntity(chapter.content)
     return chapter
   }
 }
 
-export function createEntityReadTimeline(params: ReaderParams) {
+export function createEntityReadTimeline<T = any>(params: ReaderParams) {
   const { feed, encryptionKey } = getFeedReadParams(
     params.writer,
     params.name,
     params.keyPair,
   )
-  return new Timeline({
+  return new Timeline<T>({
     bzz: params.bzz,
     feed,
     decode: createEntityTimelineDecoder({ key: encryptionKey }),
   })
 }
 
-export function createEntityTimelineLatestSubscriber(params: SubscriberParams) {
-  const timeline = createEntityReadTimeline(params)
+export function createEntityTimelineLatestSubscriber<T = any>(
+  params: SubscriberParams,
+) {
+  const timeline = createEntityReadTimeline<T>(params)
   return timeline.pollLatestChapter(params.options)
 }
 
-export function createEntityTimelineLiveSubscriber(params: SubscriberParams) {
-  const timeline = createEntityReadTimeline(params)
+export function createEntityTimelineLiveSubscriber<T = any>(
+  params: SubscriberParams,
+) {
+  const timeline = createEntityReadTimeline<T>(params)
   return timeline.live(params.options)
 }

packages/core/src/index.ts

@@ -3,6 +3,11 @@ export {
   createContactSubscriber,
   readContact,
   writeContact,
+  // Messaging
+  MailboxReaderParams,
+  MailboxWriterParams,
+  createMailboxPublisher,
+  createMailboxReader,
   // FileSystem
   FileSystemReader,
   FileSystemReaderParams,

packages/core/src/namespace.ts

@@ -17,5 +17,7 @@ export function namespace(
 
 export const CONTACT_NAME = namespace('contact')
 export const FILE_SYSTEM_NAME = namespace('fileSystem')
+export const MAILBOX_NAME = namespace('mailbox')
+export const MESSAGE_NAME = namespace('message')
 export const PEER_NAME = namespace('peer')
 export const PEER_CONTACT_NAME = namespace('peerContact')

packages/core/src/protocols/index.ts

@@ -1,4 +1,10 @@
 export { createContactSubscriber, readContact, writeContact } from './contact'
+export {
+  MailboxReaderParams,
+  MailboxWriterParams,
+  createMailboxReader,
+  createMailboxPublisher,
+} from './messaging'
 export {
   FileSystemReader,
   FileSystemReaderParams,

packages/core/src/protocols/messaging.ts

@@ -0,0 +1,36 @@
+import Bzz from '@erebos/api-bzz-base'
+
+import {
+  createEntityReadTimeline,
+  createEntityTimelinePublisher,
+} from '../channels'
+import { MAILBOX_NAME, MESSAGE_NAME } from '../namespace'
+
+interface MailboxCommonParams {
+  bzz: Bzz<any>
+  keyPair: any // TODO: keyPair type
+}
+
+export interface MailboxReaderParams extends MailboxCommonParams {
+  writer: string
+}
+
+export function createMailboxReader(params: MailboxReaderParams) {
+  return createEntityReadTimeline({
+    ...params,
+    entityType: MESSAGE_NAME,
+    name: MAILBOX_NAME,
+  })
+}
+
+export interface MailboxWriterParams extends MailboxCommonParams {
+  reader: string
+}
+
+export function createMailboxPublisher(params: MailboxWriterParams) {
+  return createEntityTimelinePublisher({
+    ...params,
+    entityType: MESSAGE_NAME,
+    name: MAILBOX_NAME,
+  })
+}

packages/core/src/schemas/contact.ts

@@ -1,11 +1,11 @@
 import { CONTACT_NAME, getID } from '../namespace'
 
-import { FileSystem, fileSystemProperty } from './fileSystem'
 import { Mailboxes, mailboxesProperty } from './messaging'
 import { Profile, profileProperty } from './profile'
+import { publicKeyProperty } from './scalars'
 
 export interface Contact {
-  files?: FileSystem
+  fileSystemKey?: string
   mailboxes?: Mailboxes
   profile?: Profile
 }
@@ -15,7 +15,7 @@ export const contactSchema = {
   $id: getID(CONTACT_NAME),
   type: 'object',
   properties: {
-    files: fileSystemProperty,
+    fileSystemKey: publicKeyProperty,
     mailboxes: mailboxesProperty,
     profile: profileProperty,
   },

packages/core/src/schemas/fileSystem.ts

@@ -1,7 +1,7 @@
 import { FILE_SYSTEM_NAME, getID } from '../namespace'
 import { EncryptionParams } from '../types'
 
-import { swarmHashProperty } from './swarmHash'
+import { swarmHashProperty } from './scalars'
 
 export interface FileEncryption extends EncryptionParams {
   key: string
@@ -29,6 +29,7 @@ export interface File extends FileMetadata {
 }
 
 export const fileProperty = {
+  type: 'object',
   required: ['hash'],
   properties: {
     hash: swarmHashProperty,

packages/core/src/schemas/messaging.ts

@@ -1,7 +1,7 @@
+import { MESSAGE_NAME, getID } from '../namespace'
+
 import { File, fileProperty } from './fileSystem'
-import { publicKeyProperty } from './publicKey'
-import { SwarmFeed, swarmFeedProperty } from './swarmFeed'
-import { swarmHashProperty } from './swarmHash'
+import { publicKeyProperty, swarmHashProperty } from './scalars'
 
 export interface MessageAttachment {
   file: File
@@ -27,6 +27,7 @@ export interface Message {
 
 export const messageSchema = {
   $async: true,
+  $id: getID(MESSAGE_NAME),
   type: 'object',
   required: ['body'],
   properties: {
@@ -41,27 +42,12 @@ export const messageSchema = {
   },
 }
 
-export interface Mailbox {
-  timeline: SwarmFeed
-  publicKey?: string
-}
-
-export const mailboxProperty = {
-  type: 'object',
-  required: ['timeline'],
-  properties: {
-    timeline: swarmFeedProperty,
-    publicKey: publicKeyProperty,
-  },
-  additionalProperties: false,
-}
-
-export type Mailboxes = Record<string, Mailbox>
+export type Mailboxes = Record<string, string>
 
 export const mailboxesProperty = {
   type: 'object',
   patternProperties: {
-    '^[0-9a-zA-Z-_. ]{1,50}$': swarmFeedProperty,
+    '^[0-9a-zA-Z-_. ]{1,50}$': publicKeyProperty,
   },
   additionalProperties: false,
 }

packages/core/src/schemas/peer.ts

@@ -1,7 +1,7 @@
 import { PEER_NAME, getID } from '../namespace'
 
 import { Profile, profileProperty } from './profile'
-import { publicKeyProperty } from './publicKey'
+import { publicKeyProperty } from './scalars'
 
 export interface Peer {
   profile?: Profile

packages/core/src/schemas/peerContact.ts

@@ -1,7 +1,6 @@
 import { PEER_CONTACT_NAME, getID } from '../namespace'
 
-import { ethereumAddressProperty } from './ethereumAddress'
-import { publicKeyProperty } from './publicKey'
+import { ethereumAddressProperty, publicKeyProperty } from './scalars'
 
 export interface PeerContact {
   contactPublicKey: string

packages/core/src/schemas/profile.ts

@@ -1,4 +1,4 @@
-import { ethereumAddressProperty } from './ethereumAddress'
+import { ethereumAddressProperty } from './scalars'
 
 export interface Profile {
   displayName?: string

packages/core/src/schemas/swarmFeed.ts → packages/core/src/schemas/scalars.ts

@@ -1,6 +1,14 @@
 import { FeedParams } from '@erebos/api-bzz-base'
 
-import { ethereumAddressProperty } from './ethereumAddress'
+export const ethereumAddressProperty = {
+  type: 'string',
+  pattern: '^0x[0-9a-f]{40}$',
+}
+
+export const publicKeyProperty = {
+  type: 'string',
+  pattern: '^[0-9a-f]{130}$',
+}
 
 export interface SwarmFeed extends FeedParams {}
 
@@ -12,3 +20,8 @@ export const swarmFeedProperty = {
     topic: { type: 'string', pattern: '^0x[0-9a-f]{64}$' },
   },
 }
+
+export const swarmHashProperty = {
+  type: 'string',
+  pattern: '^[0-9a-f]{64}$',
+}

packages/core/src/validation.ts

@@ -7,12 +7,19 @@ import { fromBuffer } from './utils'
 import { EntityPayload } from './types'
 
 import { contactSchema } from './schemas/contact'
+import { messageSchema } from './schemas/messaging'
 import { fileSystemSchema } from './schemas/fileSystem'
 import { peerSchema } from './schemas/peer'
 import { peerContactSchema } from './schemas/peerContact'
 
 const ajv = new Ajv()
-ajv.addSchema([contactSchema, fileSystemSchema, peerSchema, peerContactSchema])
+ajv.addSchema([
+  contactSchema,
+  messageSchema,
+  fileSystemSchema,
+  peerSchema,
+  peerContactSchema,
+])
 
 export async function validateEntity<T = any>(
   entity: EntityPayload<T>,