watermillchat

README

Watermill Chat Technical Demo

This package is a portfolio proof of concept of a durable HTTP live chat atop Watermill for the back-end event sourcing and Data Star for the immediate front-end rendering responsive to Server Sent Events.

Development Roadmap

If this package reaches 1.0 release, it will present a flexible and durable live chat system that can be easily integrated into projects that require chat modules with minimal dependencies and wide storage system support.

• Add message history recall on boot.
• Add random room generator.
• Make room link sharable.
• Add Olama integration plugin.
• Add user 0Auth authentication.

Installation

# Install latest version for your project:
go get -u github.com/dkotik/watermillchat@latest

# Run live demonstration on local host:
go run github.com/dkotik/watermillchat/cmd/wmcserver@latest

Documentation

Overview

Package watermillchat provides live hypermedia chat for <watermill.io> event caster. Messages are delivered as server side events.

Index

• Constants
• func Batch[T any](in <-chan T, limit int, flush time.Duration) (out chan []T)
• func ContextWithIdentity(parent context.Context, id Identity) context.Context
• type Broadcast
• type Chat
  • func New(ctx context.Context, c Configuration) (chat *Chat, err error)
  • func (c *Chat) Broadcast(ctx context.Context, b Broadcast) (err error)
  • func (c *Chat) Listen(messages <-chan *message.Message)
  • func (c *Chat) Subscribe(ctx context.Context, roomName string) <-chan []Message
• type Configuration
  • func (c Configuration) Validate() (err error)
• type HistoryConfiguration
• type HistoryRepository
• type Identity
  • func IdentityFromContext(ctx context.Context) (id Identity, ok bool)
• type Message
• type Room
  • func (r *Room) Send(ctx context.Context, m Message) error
  • func (r *Room) Subscribe(ctx context.Context) <-chan []Message
• type VoidHistoryRepository
  • func (r VoidHistoryRepository) GetRoomMessages(ctx context.Context, roomName string) ([]Message, error)
  • func (r VoidHistoryRepository) Listen(broadcasts <-chan *message.Message)
• type WatermillConfiguration

Constants

const (
	DefaultWatermillTopic             = "watermillchat"
	DefaultHistoryMostMessagesPerRoom = 1000
	DefaultHistoryDepth               = DefaultHistoryMostMessagesPerRoom // TODO: deprecate
	DefaultHistoryRetention           = time.Minute * 60 * 24             // 24 hours
	DefaultHistoryCleanupFrequency    = time.Minute * 15
	DefaultCleanupFrequency           = DefaultHistoryCleanupFrequency // TODO: deprecate
)

Functions

func Batch

func Batch[T any](
	in <-chan T,
	limit int,
	flush time.Duration,
) (out chan []T)

Batch periodically flushes incoming items as lists to outgoing channel. If list grows to limit size, it is immediately flushed.

func ContextWithIdentity

func ContextWithIdentity(parent context.Context, id Identity) context.Context

Types

type Broadcast

type Broadcast struct {
	Message
	RoomName string
}

type Chat

type Chat struct {
	// contains filtered or unexported fields
}

func New

func New(ctx context.Context, c Configuration) (chat *Chat, err error)

func (*Chat) Broadcast

func (c *Chat) Broadcast(ctx context.Context, b Broadcast) (err error)

func (*Chat) Listen

func (c *Chat) Listen(messages <-chan *message.Message)

func (*Chat) Subscribe

func (c *Chat) Subscribe(ctx context.Context, roomName string) <-chan []Message

type Configuration

type Configuration struct {
	Watermill WatermillConfiguration
	History   HistoryConfiguration
	Logger    *slog.Logger
}

func (Configuration) Validate

func (c Configuration) Validate() (err error)

type HistoryConfiguration

type HistoryConfiguration struct {
	// Defaults to [VoidHistoryRepository].
	Repository HistoryRepository

	// Retention constraints the life time of messages before their deletion.
	// Defaults to [DefaultHistoryRetention].
	Retention time.Duration

	// CleanUpFrequency is the pause between message purge
	// cycles based on retention. Defaults to [DefaultHistoryCleanupFrequency].
	CleanUpFrequency time.Duration

	// MostMessagesPerRoom constraints the maximum number of
	// returned messages per room when history is loaded using. More messages may still
	// be present in the database, if they retention duration
	// has not yet run out. Defaults to [DefaultHistoryMostMessagesPerRoom].
	MostMessagesPerRoom int
}

type HistoryRepository

type HistoryRepository interface {
	Listen(broadcasts <-chan *message.Message)
	GetRoomMessages(context.Context, string) ([]Message, error)
}

History is a list of messages grouped by room name. Must be sorted in ascending order by [Message.CreatedAt]. Apply to Chat using [WithHistory] option.

type Identity

type Identity struct {
	ID   string
	Name string
}

func IdentityFromContext

func IdentityFromContext(ctx context.Context) (id Identity, ok bool)

type Message

type Message struct {
	ID string

	// Author is the creator of the message. System messages
	// do not contain any author information.
	Author    *Identity
	Content   string
	CreatedAt int64
	UpdatedAt int64
}

type Room

type Room struct {
	// contains filtered or unexported fields
}

func (*Room) Send

func (r *Room) Send(ctx context.Context, m Message) error

func (*Room) Subscribe

func (r *Room) Subscribe(ctx context.Context) <-chan []Message

type VoidHistoryRepository

type VoidHistoryRepository struct{}

func (VoidHistoryRepository) GetRoomMessages

func (r VoidHistoryRepository) GetRoomMessages(ctx context.Context, roomName string) ([]Message, error)

func (VoidHistoryRepository) Listen

func (r VoidHistoryRepository) Listen(broadcasts <-chan *message.Message)

type WatermillConfiguration

type WatermillConfiguration struct {
	// Topic where the messages are published to and read from.
	// Defaults to [DefaultWatermillTopic].
	Topic      string
	Publisher  message.Publisher
	Subscriber message.Subscriber
}