Import chat2 example

2021-06-09 15:27:37 +08:00 · 2021-06-09 15:27:37 +08:00 · 5319078504
parent 08c55e1743
commit 5319078504
10 changed files with 1963 additions and 0 deletions
--- a/examples/chat2/Makefile
+++ b/examples/chat2/Makefile
@ -0,0 +1,6 @@
+.PHONY: all build
+
+build:
+	go build -o build/chat2 .
+
+all: build
--- a/examples/chat2/README.md
+++ b/examples/chat2/README.md
@ -0,0 +1,63 @@
+# Using the `chat2` application
+
+## Background
+
+The `chat2` application is a basic command-line chat app using the [Waku v2 suite of protocols](https://specs.vac.dev/specs/waku/v2/waku-v2). It connects to a [fleet of test nodes](fleets.status.im) to provide end-to-end p2p chat capabilities. The Waku team is currently using this application for internal testing. If you want try our protocols, or join the dogfooding fun, follow the instructions below.
+
+## Preparation
+```
+make
+```
+
+## Basic application usage
+
+To start the `chat2` application in its most basic form, run the following from the project directory
+
+```
+./build/chat2
+```
+
+The app will randomly select and connect to a peer from the test fleet.
+
+```
+No static peers configured. Choosing one at random from test fleet...
+```
+
+Wait for the chat prompt (`>`) and chat away!
+
+## Retrieving historical messages
+
+TODO
+
+## Specifying a static peer
+
+In order to connect to a *specific* node as [`relay`](https://specs.vac.dev/specs/waku/v2/waku-relay) peer, define that node's `multiaddr` as a `staticnode` when starting the app:
+
+```
+./build/chat2 -staticnode=/ip4/134.209.139.210/tcp/30303/p2p/16Uiu2HAmPLe7Mzm8TsYUubgCAW1aJoeFScxrLj8ppHFivPo97bUZ
+```
+
+This will bypass the random peer selection process and connect to the specified node.
+
+## In-chat options
+
+| Command | Effect |
+| --- | --- |
+| `/help` | displays available in-chat commands |
+| `/connect` | interactively connect to a new peer |
+| `/nick` | change nickname for current chat session |
+| `/peers` | Display the list of connected peers |
+
+## `chat2` message protobuf format
+
+Each `chat2` message is encoded as follows
+
+```protobuf
+message Chat2Message {
+  uint64 timestamp = 1;
+  string nick = 2;
+  bytes payload = 3;
+}
+```
+
+where `timestamp` is the Unix timestamp of the message, `nick` is the relevant `chat2` user's selected nickname and `payload` is the actual chat message being sent. The `payload` is the byte array representation of a UTF8 encoded string.
--- a/examples/chat2/build/.gitignore
+++ b/examples/chat2/build/.gitignore
@ -0,0 +1,2 @@
+*
+!.gitignore
--- a/examples/chat2/chat.go
+++ b/examples/chat2/chat.go
@ -0,0 +1,145 @@
+package main
+
+import (
+	"chat2/pb"
+	"context"
+	"crypto/sha256"
+	"time"
+
+	"github.com/golang/protobuf/proto"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/status-im/go-waku/waku/v2/node"
+	wpb "github.com/status-im/go-waku/waku/v2/protocol/pb"
+	"golang.org/x/crypto/pbkdf2"
+)
+
+// Chat represents a subscription to a single PubSub topic. Messages
+// can be published to the topic with Chat.Publish, and received
+// messages are pushed to the Messages channel.
+type Chat struct {
+	// Messages is a channel of messages received from other peers in the chat room
+	Messages chan *pb.Chat2Message
+
+	sub  *node.Subscription
+	node *node.WakuNode
+
+	self         peer.ID
+	contentTopic string
+	useV1Payload bool
+	nick         string
+}
+
+// NewChat tries to subscribe to the PubSub topic for the room name, returning
+// a ChatRoom on success.
+func NewChat(n *node.WakuNode, selfID peer.ID, contentTopic string, useV1Payload bool, nickname string) (*Chat, error) {
+	// join the default waku topic and subscribe to it
+	sub, err := n.Subscribe(nil)
+	if err != nil {
+		return nil, err
+	}
+
+	c := &Chat{
+		node:         n,
+		sub:          sub,
+		self:         selfID,
+		contentTopic: contentTopic,
+		nick:         nickname,
+		useV1Payload: useV1Payload,
+		Messages:     make(chan *pb.Chat2Message, 1024),
+	}
+
+	// start reading messages from the subscription in a loop
+	go c.readLoop()
+
+	return c, nil
+}
+
+func generateSymKey(password string) []byte {
+	// AesKeyLength represents the length (in bytes) of an private key
+	AESKeyLength := 256 / 8
+	return pbkdf2.Key([]byte(password), nil, 65356, AESKeyLength, sha256.New)
+}
+
+// Publish sends a message to the pubsub topic.
+func (cr *Chat) Publish(ctx context.Context, message string) error {
+
+	msg := &pb.Chat2Message{
+		Timestamp: uint64(time.Now().Unix()),
+		Nick:      cr.nick,
+		Payload:   []byte(message),
+	}
+
+	msgBytes, err := proto.Marshal(msg)
+	if err != nil {
+		return err
+	}
+
+	var version uint32
+	var timestamp float64 = float64(time.Now().UnixNano())
+	var keyInfo *node.KeyInfo = &node.KeyInfo{}
+
+	if cr.useV1Payload { // Use WakuV1 encryption
+		keyInfo.Kind = node.Symmetric
+		keyInfo.SymKey = generateSymKey(cr.contentTopic)
+		version = 1
+	} else {
+		keyInfo.Kind = node.None
+		version = 0
+	}
+
+	p := new(node.Payload)
+	p.Data = msgBytes
+	p.Key = keyInfo
+
+	payload, err := p.Encode(version)
+	if err != nil {
+		return err
+	}
+
+	wakuMsg := &wpb.WakuMessage{
+		Payload:      payload,
+		Version:      version,
+		ContentTopic: cr.contentTopic,
+		Timestamp:    timestamp,
+	}
+
+	_, err = cr.node.Publish(ctx, wakuMsg, nil)
+
+	return err
+}
+
+func (cr *Chat) decodeMessage(wakumsg *wpb.WakuMessage) {
+	var keyInfo *node.KeyInfo = &node.KeyInfo{}
+	if cr.useV1Payload { // Use WakuV1 encryption
+		keyInfo.Kind = node.Symmetric
+		keyInfo.SymKey = generateSymKey(cr.contentTopic)
+	} else {
+		keyInfo.Kind = node.None
+	}
+
+	payload, err := node.DecodePayload(wakumsg, keyInfo)
+	if err != nil {
+		return
+	}
+
+	msg := &pb.Chat2Message{}
+	if err := proto.Unmarshal(payload.Data, msg); err != nil {
+		return
+	}
+
+	// send valid messages onto the Messages channel
+	cr.Messages <- msg
+}
+
+// readLoop pulls messages from the pubsub topic and pushes them onto the Messages channel.
+func (cr *Chat) readLoop() {
+	for value := range cr.sub.C {
+		cr.decodeMessage(value.Message())
+	}
+}
+
+func (cr *Chat) displayMessages(messages []*wpb.WakuMessage) {
+	for _, msg := range messages {
+		cr.decodeMessage(msg)
+	}
+}
--- a/examples/chat2/go.mod
+++ b/examples/chat2/go.mod
@ -0,0 +1,17 @@
+module chat2
+
+go 1.15
+
+replace github.com/status-im/go-waku => ../..
+
+require (
+	github.com/ethereum/go-ethereum v1.10.1
+	github.com/gdamore/tcell/v2 v2.2.0
+	github.com/golang/protobuf v1.4.3
+	github.com/ipfs/go-log v1.0.4
+	github.com/libp2p/go-libp2p-core v0.8.5
+	github.com/rivo/tview v0.0.0-20210312174852-ae9464cc3598
+	github.com/status-im/go-waku v0.0.0-20210428201044-3d8aae5b81b9
+	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9
+	google.golang.org/protobuf v1.25.0
+)
--- a/examples/chat2/go.sum
+++ b/examples/chat2/go.sum
--- a/examples/chat2/main.go
+++ b/examples/chat2/main.go
@ -0,0 +1,224 @@
+package main
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"log"
+	mrand "math/rand"
+	"net"
+	"net/http"
+	"os"
+	"time"
+
+	"github.com/ethereum/go-ethereum/crypto"
+	logging "github.com/ipfs/go-log"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/status-im/go-waku/waku/v2/node"
+	"github.com/status-im/go-waku/waku/v2/protocol/store"
+)
+
+var DefaultContentTopic string = "/toy-chat/2/huilong/proto"
+
+func main() {
+	mrand.Seed(time.Now().UTC().UnixNano())
+
+	nickFlag := flag.String("nick", "", "nickname to use in chat. will be generated if empty")
+	fleetFlag := flag.String("fleet", "wakuv2.prod", "Select the fleet to connect to. (wakuv2.prod, wakuv2.test)")
+	contentTopicFlag := flag.String("contenttopic", DefaultContentTopic, "content topic to use for the chat")
+	nodeKeyFlag := flag.String("nodekey", "", "private key for this node. will be generated if empty")
+	staticNodeFlag := flag.String("staticnode", "", "connects to a node. will get a random node from fleets.status.im if empty")
+	storeNodeFlag := flag.String("storenode", "", "connects to a store node to retrieve messages. will get a random node from fleets.status.im if empty")
+	port := flag.Int("port", 0, "port. Will be random if 0")
+	payloadV1Flag := flag.Bool("payloadV1", false, "use Waku v1 payload encoding/encryption. default false")
+
+	flag.Parse()
+
+	hostAddr, _ := net.ResolveTCPAddr("tcp", fmt.Sprintf("0.0.0.0:%d", *port))
+
+	if *fleetFlag != "wakuv2.prod" && *fleetFlag != "wakuv2.test" {
+		fmt.Println("Invalid fleet. Valid values are wakuv2.prod and wakuv2.test")
+		return
+	}
+
+	// use the nickname from the cli flag, or a default if blank
+	nodekey := *nodeKeyFlag
+	if len(nodekey) == 0 {
+		var err error
+		nodekey, err = randomHex(32)
+		if err != nil {
+			fmt.Println("Could not generate random key")
+			return
+		}
+	}
+	prvKey, err := crypto.HexToECDSA(nodekey)
+
+	ctx := context.Background()
+	wakuNode, err := node.New(ctx,
+		node.WithPrivateKey(prvKey),
+		node.WithHostAddress([]net.Addr{hostAddr}),
+		node.WithWakuRelay(),
+		node.WithWakuStore(false),
+	)
+	if err != nil {
+		fmt.Print(err)
+		return
+	}
+
+	// use the nickname from the cli flag, or a default if blank
+	nick := *nickFlag
+	if len(nick) == 0 {
+		nick = defaultNick(wakuNode.Host().ID())
+	}
+
+	// join the chat
+	chat, err := NewChat(wakuNode, wakuNode.Host().ID(), *contentTopicFlag, *payloadV1Flag, nick)
+	if err != nil {
+		panic(err)
+	}
+
+	// Display panic level to reduce log noise
+	lvl, err := logging.LevelFromString("panic")
+	if err != nil {
+		panic(err)
+	}
+	logging.SetAllLoggers(lvl)
+
+	ui := NewChatUI(ctx, chat)
+
+	// Connect to a static node or use random node from fleets.status.im
+	go func() {
+		time.Sleep(200 * time.Millisecond)
+
+		staticnode := *staticNodeFlag
+		storenode := *storeNodeFlag
+
+		var fleetData []byte
+		if len(staticnode) == 0 || len(storenode) == 0 {
+			fleetData = getFleetData()
+		}
+
+		if len(staticnode) == 0 {
+			ui.displayMessage(fmt.Sprintf("No static peers configured. Choosing one at random from %s fleet...", *fleetFlag))
+			staticnode = getRandomFleetNode(fleetData, *fleetFlag)
+		}
+
+		err = wakuNode.DialPeer(staticnode)
+		if err != nil {
+			ui.displayMessage("Could not connect to peer: " + err.Error())
+			return
+		} else {
+			ui.displayMessage("Connected to peer: " + staticnode)
+
+		}
+
+		if len(storenode) == 0 {
+			ui.displayMessage(fmt.Sprintf("No store node configured. Choosing one at random from %s fleet...", *fleetFlag))
+			storenode = getRandomFleetNode(fleetData, *fleetFlag)
+		}
+
+		storeNodeId, err := wakuNode.AddStorePeer(storenode)
+		if err != nil {
+			ui.displayMessage("Could not connect to storenode: " + err.Error())
+			return
+		} else {
+			ui.displayMessage("Connected to storenode: " + storenode)
+		}
+
+		time.Sleep(300 * time.Millisecond)
+		ui.displayMessage("Querying historic messages")
+
+		tCtx, _ := context.WithTimeout(ctx, 5*time.Second)
+		response, err := wakuNode.Query(tCtx, []string{*contentTopicFlag}, 0, 0,
+			store.WithAutomaticRequestId(),
+			store.WithPeer(*storeNodeId),
+			store.WithPaging(true, 0))
+
+		if err != nil {
+			ui.displayMessage("Could not query storenode: " + err.Error())
+		} else {
+			chat.displayMessages(response.Messages)
+		}
+	}()
+
+	//draw the UI
+	if err = ui.Run(); err != nil {
+		printErr("error running text UI: %s", err)
+	}
+}
+
+// Generates a random hex string with a length of n
+func randomHex(n int) (string, error) {
+	bytes := make([]byte, n)
+	if _, err := rand.Read(bytes); err != nil {
+		return "", err
+	}
+	return hex.EncodeToString(bytes), nil
+}
+
+// printErr is like fmt.Printf, but writes to stderr.
+func printErr(m string, args ...interface{}) {
+	fmt.Fprintf(os.Stderr, m, args...)
+}
+
+// defaultNick generates a nickname based on the $USER environment variable and
+// the last 8 chars of a peer ID.
+func defaultNick(p peer.ID) string {
+	return fmt.Sprintf("%s-%s", os.Getenv("USER"), shortID(p))
+}
+
+// shortID returns the last 8 chars of a base58-encoded peer id.
+func shortID(p peer.ID) string {
+	pretty := p.Pretty()
+	return pretty[len(pretty)-8:]
+}
+
+func getFleetData() []byte {
+	url := "https://fleets.status.im"
+	httpClient := http.Client{
+		Timeout: time.Second * 2,
+	}
+
+	req, err := http.NewRequest(http.MethodGet, url, nil)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	res, getErr := httpClient.Do(req)
+	if getErr != nil {
+		log.Fatal(getErr)
+	}
+
+	if res.Body != nil {
+		defer res.Body.Close()
+	}
+
+	body, readErr := ioutil.ReadAll(res.Body)
+	if readErr != nil {
+		log.Fatal(readErr)
+	}
+
+	return body
+}
+
+func getRandomFleetNode(data []byte, fleetId string) string {
+	var result map[string]interface{}
+	json.Unmarshal(data, &result)
+	fleets := result["fleets"].(map[string]interface{})
+	fleet := fleets[fleetId].(map[string]interface{})
+	waku := fleet["waku"].(map[string]interface{})
+
+	var wakunodes []string
+	for v := range waku {
+		wakunodes = append(wakunodes, v)
+		break
+	}
+
+	randKey := wakunodes[mrand.Intn(len(wakunodes))]
+
+	return waku[randKey].(string)
+}
--- a/examples/chat2/pb/chat2.pb.go
+++ b/examples/chat2/pb/chat2.pb.go
@ -0,0 +1,165 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.25.0
+// 	protoc        v3.14.0
+// source: chat2.proto
+
+package pb
+
+import (
+	proto "github.com/golang/protobuf/proto"
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+// This is a compile-time assertion that a sufficiently up-to-date version
+// of the legacy proto package is being used.
+const _ = proto.ProtoPackageIsVersion4
+
+type Chat2Message struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Timestamp uint64 `protobuf:"varint,1,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
+	Nick      string `protobuf:"bytes,2,opt,name=nick,proto3" json:"nick,omitempty"`
+	Payload   []byte `protobuf:"bytes,3,opt,name=payload,proto3" json:"payload,omitempty"`
+}
+
+func (x *Chat2Message) Reset() {
+	*x = Chat2Message{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_chat2_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *Chat2Message) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Chat2Message) ProtoMessage() {}
+
+func (x *Chat2Message) ProtoReflect() protoreflect.Message {
+	mi := &file_chat2_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use Chat2Message.ProtoReflect.Descriptor instead.
+func (*Chat2Message) Descriptor() ([]byte, []int) {
+	return file_chat2_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *Chat2Message) GetTimestamp() uint64 {
+	if x != nil {
+		return x.Timestamp
+	}
+	return 0
+}
+
+func (x *Chat2Message) GetNick() string {
+	if x != nil {
+		return x.Nick
+	}
+	return ""
+}
+
+func (x *Chat2Message) GetPayload() []byte {
+	if x != nil {
+		return x.Payload
+	}
+	return nil
+}
+
+var File_chat2_proto protoreflect.FileDescriptor
+
+var file_chat2_proto_rawDesc = []byte{
+	0x0a, 0x0b, 0x63, 0x68, 0x61, 0x74, 0x32, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x02, 0x70,
+	0x62, 0x22, 0x5a, 0x0a, 0x0c, 0x43, 0x68, 0x61, 0x74, 0x32, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67,
+	0x65, 0x12, 0x1c, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12,
+	0x12, 0x0a, 0x04, 0x6e, 0x69, 0x63, 0x6b, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e,
+	0x69, 0x63, 0x6b, 0x12, 0x18, 0x0a, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18, 0x03,
+	0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x62, 0x06, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_chat2_proto_rawDescOnce sync.Once
+	file_chat2_proto_rawDescData = file_chat2_proto_rawDesc
+)
+
+func file_chat2_proto_rawDescGZIP() []byte {
+	file_chat2_proto_rawDescOnce.Do(func() {
+		file_chat2_proto_rawDescData = protoimpl.X.CompressGZIP(file_chat2_proto_rawDescData)
+	})
+	return file_chat2_proto_rawDescData
+}
+
+var file_chat2_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
+var file_chat2_proto_goTypes = []interface{}{
+	(*Chat2Message)(nil), // 0: pb.Chat2Message
+}
+var file_chat2_proto_depIdxs = []int32{
+	0, // [0:0] is the sub-list for method output_type
+	0, // [0:0] is the sub-list for method input_type
+	0, // [0:0] is the sub-list for extension type_name
+	0, // [0:0] is the sub-list for extension extendee
+	0, // [0:0] is the sub-list for field type_name
+}
+
+func init() { file_chat2_proto_init() }
+func file_chat2_proto_init() {
+	if File_chat2_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_chat2_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*Chat2Message); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_chat2_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   1,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_chat2_proto_goTypes,
+		DependencyIndexes: file_chat2_proto_depIdxs,
+		MessageInfos:      file_chat2_proto_msgTypes,
+	}.Build()
+	File_chat2_proto = out.File
+	file_chat2_proto_rawDesc = nil
+	file_chat2_proto_goTypes = nil
+	file_chat2_proto_depIdxs = nil
+}
--- a/examples/chat2/pb/chat2.proto
+++ b/examples/chat2/pb/chat2.proto
@ -0,0 +1,9 @@
+syntax = "proto3";
+
+package pb;
+
+message Chat2Message {
+    uint64 timestamp = 1;
+    string nick = 2;
+    bytes payload = 3;
+}
--- a/examples/chat2/ui.go
+++ b/examples/chat2/ui.go
@ -0,0 +1,212 @@
+package main
+
+import (
+	"chat2/pb"
+	"context"
+	"fmt"
+	"io"
+	"strings"
+	"time"
+
+	"github.com/gdamore/tcell/v2"
+	"github.com/rivo/tview"
+	"github.com/status-im/go-waku/waku/v2/protocol/relay"
+)
+
+// ChatUI is a Text User Interface (TUI) for a ChatRoom.
+// The Run method will draw the UI to the terminal in "fullscreen"
+// mode. You can quit with Ctrl-C, or by typing "/quit" into the
+// chat prompt.
+type ChatUI struct {
+	app  *tview.Application
+	chat *Chat
+
+	msgW    io.Writer
+	inputCh chan string
+	doneCh  chan struct{}
+
+	ctx context.Context
+}
+
+// NewChatUI returns a new ChatUI struct that controls the text UI.
+// It won't actually do anything until you call Run().
+func NewChatUI(ctx context.Context, chat *Chat) *ChatUI {
+	chatUI := new(ChatUI)
+
+	app := tview.NewApplication()
+
+	// make a NewChatUI text view to contain our chat messages
+	msgBox := tview.NewTextView()
+	msgBox.SetDynamicColors(true)
+	msgBox.SetBorder(true)
+	msgBox.SetTitle("chat2 example")
+
+	// text views are io.Writers, but they don't automatically refresh.
+	// this sets a change handler to force the app to redraw when we get
+	// new messages to display.
+	msgBox.SetChangedFunc(func() {
+		app.Draw()
+	})
+
+	// an input field for typing messages into
+	inputCh := make(chan string, 32)
+	input := tview.NewInputField().
+		SetLabel(chat.nick + " > ").
+		SetFieldWidth(0).
+		SetFieldBackgroundColor(tcell.ColorBlack)
+
+	// the done func is called when the user hits enter, or tabs out of the field
+	input.SetDoneFunc(func(key tcell.Key) {
+		if key != tcell.KeyEnter {
+			// we don't want to do anything if they just tabbed away
+			return
+		}
+		line := input.GetText()
+
+		if len(line) == 0 {
+			// ignore blank lines
+			return
+		}
+
+		input.SetText("")
+
+		// bail if requested
+		if line == "/quit" {
+			app.Stop()
+			return
+		}
+
+		// add peer
+		if strings.HasPrefix(line, "/connect ") {
+			peer := strings.TrimPrefix(line, "/connect ")
+			go func(peer string) {
+				chatUI.displayMessage("Connecting to peer...")
+				err := chat.node.DialPeer(peer)
+				if err != nil {
+					chatUI.displayMessage(err.Error())
+				} else {
+					chatUI.displayMessage("Peer connected succesfully")
+				}
+			}(peer)
+			return
+		}
+
+		// list peers
+		if line == "/peers" {
+			peers := chat.node.Relay().PubSub().ListPeers(string(relay.DefaultWakuTopic))
+			if len(peers) == 0 {
+				chatUI.displayMessage("No peers available")
+			}
+			for _, p := range peers {
+				chatUI.displayMessage("- " + p.Pretty())
+			}
+			return
+		}
+
+		// change nick
+		if strings.HasPrefix(line, "/nick ") {
+			newNick := strings.TrimSpace(strings.TrimPrefix(line, "/nick "))
+			chat.nick = newNick
+			input.SetLabel(chat.nick + " > ")
+			return
+		}
+
+		if line == "/help" {
+			chatUI.displayMessage(`
+Available commands:
+/connect multiaddress - dials a node adding it to the list of connected peers
+/peers - list of peers connected to this node
+/nick newNick - change the user's nickname
+/quit - closes the app
+`)
+			return
+		}
+
+		// send the line onto the input chan and reset the field text
+		inputCh <- line
+	})
+
+	chatPanel := tview.NewFlex().
+		AddItem(msgBox, 0, 1, false)
+
+	// flex is a vertical box with the chatPanel on top and the input field at the bottom.
+	flex := tview.NewFlex().
+		SetDirection(tview.FlexRow).
+		AddItem(chatPanel, 0, 1, false).
+		AddItem(input, 1, 1, true)
+
+	app.SetRoot(flex, true)
+
+	chatUI.app = app
+	chatUI.msgW = msgBox
+	chatUI.chat = chat
+	chatUI.ctx = ctx
+	chatUI.inputCh = inputCh
+	chatUI.doneCh = make(chan struct{}, 1)
+
+	for _, addr := range chat.node.ListenAddresses() {
+		chatUI.displayMessage(fmt.Sprintf("Listening on %s", addr))
+	}
+
+	return chatUI
+}
+
+// Run starts the chat event loop in the background, then starts
+// the event loop for the text UI.
+func (ui *ChatUI) Run() error {
+	ui.displayMessage("\nWelcome, " + ui.chat.nick)
+	ui.displayMessage("type /help to see available commands \n")
+
+	go ui.handleEvents()
+	defer ui.end()
+
+	return ui.app.Run()
+}
+
+// end signals the event loop to exit gracefully
+func (ui *ChatUI) end() {
+	ui.doneCh <- struct{}{}
+}
+
+// displayChatMessage writes a ChatMessage from the room to the message window,
+// with the sender's nick highlighted in green.
+func (ui *ChatUI) displayChatMessage(cm *pb.Chat2Message) {
+	t := time.Unix(int64(cm.Timestamp), 0)
+	prompt := withColor("green", fmt.Sprintf("<%s> %s:", t.Format("Jan 02, 15:04"), cm.Nick))
+	fmt.Fprintf(ui.msgW, "%s %s\n", prompt, cm.Payload)
+}
+
+// displayMessage writes a blue message to output
+func (ui *ChatUI) displayMessage(msg string) {
+	fmt.Fprintf(ui.msgW, "%s\n", withColor("grey", msg))
+}
+
+// handleEvents runs an event loop that sends user input to the chat room
+// and displays messages received from the chat room. It also periodically
+// refreshes the list of peers in the UI.
+func (ui *ChatUI) handleEvents() {
+	for {
+		select {
+		case input := <-ui.inputCh:
+			err := ui.chat.Publish(ui.ctx, input)
+			if err != nil {
+				printErr("publish error: %s", err)
+			}
+
+		case m := <-ui.chat.Messages:
+			// when we receive a message from the chat room, print it to the message window
+			ui.displayChatMessage(m)
+
+		case <-ui.ctx.Done():
+			return
+
+		case <-ui.doneCh:
+			return
+		}
+	}
+}
+
+// withColor wraps a string with color tags for display in the messages text box.
+func withColor(color, msg string) string {
+	return fmt.Sprintf("[%s]%s[-]", color, msg)
+}