Help getting started

[gostega]

Hi, thanks so much for doing this package. It's just what I need at the point I'm at in my project.

I copied from the complex and simple examples.

I have this much working

UPDATE: made sure it works, here's a proof of concept https://github.com/gostega/go-sse-poc

main.go

package main

import (
	"fmt"
	"io"
	"log"
	"net/http"
	"surfactor/pkg/controllers"
	"surfactor/pkg/models"
	"text/template"
	"time"

	"github.com/labstack/echo/v4"
	"github.com/labstack/echo/v4/middleware"
	"github.com/tmaxmax/go-sse"
)

const (
	topicRandomNumbers = "numbers"
	topicStateChanges  = "evt_statechange"
)

var sseHandler = &sse.Server{
	Provider: &sse.Joe{
		ReplayProvider: &sse.ValidReplayProvider{
			TTL:        time.Minute * 5,
			GCInterval: time.Minute,
			AutoIDs:    false,
		},
	},
	Logger: nil,
	OnSession: func(s *sse.Session) (sse.Subscription, bool) {
		topics := s.Req.URL.Query()["topic"]
		for _, topic := range topics {
			if topic != topicRandomNumbers && topic != topicStateChanges {
				fmt.Fprintf(s.Res, "invalid topic %q; supported are %q, %q", topic, topicRandomNumbers, topicStateChanges)
				s.Res.WriteHeader(http.StatusBadRequest)
				return sse.Subscription{}, false
			}
		}
		if len(topics) == 0 {
			// Provide default topics, if none are given.
			topics = []string{topicRandomNumbers, topicStateChanges}
		}

		return sse.Subscription{
			Client:      s,
			LastEventID: s.LastEventID,
			Topics:      append(topics, sse.DefaultTopic), // the shutdown message is sent on the default topic
		}, true
	},
}

func sseTest(c echo.Context) error {
	// s := &sse.Server{}

	// seededRand := rand.New(rand.NewSource(time.Now().UnixNano()))
	// const charset = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
	go func() {

		// randID := seededRand.Intn(10)
		// b := make([]byte, 10)

		// for range time.Tick(time.Second) {
		// 	m := &sse.Message{}
		// 	m.ID = sse.ID(fmt.Sprintf("thread:%d", randID))
		// 	for i := range b {
		// 		b[i] = charset[seededRand.Intn(len(charset))]
		// 	}

		// 	m.AppendData(string(b))
		// 	_ = sseHandler.Publish(m)
		// }
		m := &sse.Message{}
		m.ID = sse.ID(topicStateChanges)
		m.AppendData("Connected")
		_ = sseHandler.Publish(m)

	}()

	sseHandler.ServeHTTP(c.Response(), c.Request())

	return nil
}

func main() {

        // -------------------------------------
	// ECHO ROUTER INITIALISATION
	// -------------------------------------
	e := echo.New()
	e.Use(middleware.Logger())
	// e.Use(middleware.Recover()) // only works with panics, not 'fatal'
	e.Renderer = NewTemplates()

       e.GET("/events", sseTest)

	/* Start the http server */
	go e.Logger.Fatal(e.Start(port))

        // would also be nice to send a closing message here. "bye" or something
	// I tried this but nothing happens. Perhaps it only works if my app gets terminated differently? I'm just using CTRL+C
	mBye := &sse.Message{Type: sse.Type("close")}
	mBye.ID = sse.ID(topicStateChanges)
	mBye.AppendData("Bye")
	defer sseHandler.Publish(mBye)
}

It works, if I have the random string code uncommented, it will send me random strings every second.
If I have it commented, like above, then I have only the 'Connected' message which I did just to show that the connection was successful.

$ curl localhost:9090/events?topic=evt_statechange
id: evt_statechange
data: Connected

The part that I'm stuck on now, is publishing messages from elsewhere in my code. I have other modules that do things like update the state on a node. And interact with a database. How do I call `_ = sseHandler.Publish("mymessage") from elsewhere?

For example I have another file models/node.go

package models

type Node struct {
	gorm.Model
	Name        string `json:"name"`
	State       State  `json:"state"`
	Target      string `json:"target"`
	Description string `json:"description"`
	Parents     []Path `gorm:"many2many:nodes_paths;"`
	fromState   State  //making the variable start with lowercase makes it invisible to json
}

func (n *Node) AfterUpdate(tx *gorm.DB) (err error) {
	if n.State != n.fromState {
		// how do I from here publish a message to the client?
	}
}

[gostega]

I sort of got it working by moving it into a sub package. Don't know if this is good Go practice or styles. If anyone has anything better, please share

https://github.com/gostega/go-sse-poc/pull/1/files

[tmaxmax]

Sending a close message

So, to solve the first issue – of sending a "close" message – you can look at the complex example. It's not using Echo but the standard library's HTTP library but I believe it should be adaptable by looking at Echo's documentation. Personally I don't have any Echo experience so I'm not able to help you with precise directions on that. Here is the example – the net/http.Server.RegisterOnShutdown function does the trick.

Quick fix to dispatching events after update

For the GORM issue, an easy – but I would say dirty – solution would be the statement context. When making a GORM query, add the context as described here, but use a context which has the SSE handler as a value. For example:

db.WithContext(context.WithValue(ctx, "sseHandler", sseHandler)).Find(...)

where sseHandler is of type sse.Server and ctx is a context previously obtained (by means of an http.Request or just a context.Background(). Do not use a string key, as I did in the example. Read the context.WithValue docs for information.

This is one way to avoid creating a package for the SSE handler and having that as some sort of singleton – which I would consider an antipattern which gives away code design issues. Read further for an elaboration on that.

Some code design stuff

Now, for what I would consider a cleaner solution. Normally, to handle a request, a backend server is structured in the following manner:

1. the route handlers: this does user input deserialization (e.g. JSON to struct), calls into the business logic layer with the deserialized input, serializes the result of the business logic layer into the wire format (HTML, JSON etc.) and sends it to the client, all while handling any errors which occurred in the process.
2. the business logic layer: this first validates the user input, and then does various operations on the data source(s) (a database, for example) and other services (for example, the server-sent events handler) in order to execute a single business operation; you may choose to omit explicitly creating this layer for simple services (which may very well be your case).
3. the data layer: this usually does the basic CRUD operations on the data, abstracting away database-specific details, with which the business layer should not be concerned.

Let's run through an example: posting a tweet. Here's what the business requirements are:

• a tweet should be not empty and not longer than 120 characters (we'll do validation)
• posted tweets can be subsequently viewed forever (in other words, stored in the database)
• users which follow the poster should be notified of the new post (to use the SSE handler)
  This will be a JSON API, so both the user input and the server response will be in JSON.

At the code level, to each of these layers corresponds a struct with methods. For example, the data layer may look like this (I'll be using some pseudocodish Go which ignores imports for the examples):

package gormdata // assume the file path for this is data/gormdata/repository.go, it will make sense later

type Repository struct {
    db *gorm.DB
}

func (r *Repository) CreateTweet(ctx context.Context, user domain.User, content domain.TweetContent) (domain.Tweet, error)
func (r *Repository) GetUser(ctx context.Context, id domain.ID) (domain.User, error)
func (r *Repository) GetFollowers(ctx context.Context, user domain.User) ([]domain.User, error)

The methods above would implement the queries necessary to do what their names describe. Note that there is no sseHandler involved at this level – the data layer is only concerned with data. As an implementation detail, we still take a context.Context in even though we don't use it to pass the sseHandler to the hooks (we won't be using any GORM hooks anyway). The context is useful as a cancellation signal – that is, to stop the database queries/transaction if, for example, the user cancelled the request, the database timed out (in other words, set a limit on how long a request can take) and so on. If using GORM, always use db.WithContext for each request and immediately check the error after querying. Something like this:

if err := db.WithContext(ctx).Find(...).Error; err != nil {
    return err // or other appropriate handling
}

Anyway, now that the data layer is in place, we should create the business logic/domain layer. Here are the models:

package domain

type ID uuid.UUID // assume some external UUID package; could also be int64 or whatever else

type User struct {
    ID ID
    // other fields, for example a name etc.
}

// TweetContent is a string which can be the content of a tweet.
// That is, a text of length greater than 0 and smaller or equal to 120.
type TweetContent struct {
    text string
}

// String returns the inner text as a string.
func (s TweetContent) String() string

// NewTweetContent validates that the input is not empty and not longer than 120 characters
// and returns a TweetContent which contains it. An error is otherwise returned.
func NewTweetContent(input string) (TweetContent, error)

type Tweet struct {
    ID ID
    Poster *model.User // or maybe just posterID
    Content TweetContent
    // other fields, for example a createdAt would be useful – not included here
    // because it's not specifically stated in our business requirements above.
} 

Something to note is that we do not consider these models to be GORM-specific here. We aren't and won't be making use of any GORM-specific features – the whole point is to be decoupled from the database outside the data layer. GORM models should probably be inside the gormdata package.

As a side note: By creating the TweetContent type instead of just using string we can ensure at the type level that anywhere where a valid tweet content is required a valid tweet content is provided (see Repository.CreateTweet). The string type does not encode the constraints the tweet content has, but this type, because it cannot be created through any other means than the constructor function, does. In general you want to try as much as possible to ensure validity – and generally hold as much information as possible about a value – at the type level, so you get help from the compiler.

Now comes our business logic handler in the same package:

package domain

type Repository interface {
    CreateTweet(ctx context.Context, user User, content TweetContent) (Tweet, error)
    GetUser(ctx context.Context, id ID) (User, error)
    GetFollowers(ctx context.Context, user User) ([]User, error)
}

type Domain struct {
    repo Repository
    sseHandler *sse.Server
}

func (d *Domain) PostTweet(ctx context.Context, userID, content string) (Tweet, error)

Note that we define the Repository interface here. The domain package defines how the data layer should look like, as it is the one that uses it – the consumer. Read the Go code review comments section on interfaces for more details on this (also, I consider the Go code review comments to be, as a whole, an essential must read for every Go programmer, so make sure to go through it fully if you haven't). By not receiving the gormdata.Repository directly we:

1. don't concern ourselves (much, transactions should probably be implemented somehow) anymore with database details in the implementation layer, which simplifies and streamlines the code and makes changes easier down the line (for example, if the database schema changes and the queries become more complex, you just do the change inside the data layer, not everywhere the specific query is used – so this also helps with preventing code duplication), and
2. open up the possibility to switch the database/ORM in the future. For example, you are bound to find out that GORM is not the best choice for interacting with the database (and that ORMs are not the first choice in Go); by having this separation you won't have to rewrite your entire application, in order to get rid of the models and disentagle the hooks code, but instead all you'll have to do is create a new package with a new repository and swap them later.
   What we've done here is basically dependency injection.

Anyway, here's what the PostTweet function will do:

• create a domain.ID from the userID string and a TweetContent from the content string; return errors if any fails
• get the user's followers using Repository.GetFollowers; we do this before creating the tweet itself so, in case getting the followers fails, no tweet is created (we must do this because we haven't defined a way to do database transactions at the data layer – this is an example and in the real world you will want to have this)
• create the tweet, using Repository.GetUser and Repository.CreateTweet; return errors if any fails
• build the server-sent event and dispatch it to each of the followers; we'll use the followers' user IDs as message topics so only the respective users are reached (this assumes that each of those users is subscribed to a topic with their user ID as a name; there are better ways to do this, for example to have them subscribed to a topic that corresponds to the poster, but for this example it's simpler this way); return any errors, if they occur, because if a sse.Server.Publish error occurrs the handler is closed and the request should fail anyway

This would be the business logic layer. Now, for the final layer, to route handlers:

package handler

type Handler struct {
    domain *domain.Domain
}

func (h *Handler) PostTweet(w http.ResponseWriter, r *http.Request) // this method is an http.HandlerFunc, as it will be passed directly to the router. You can use the interface expected by Echo if you're going to use it.

type postTweetInput struct {
    UserID string `json:"user_id"`
    Content string `json:"content"`
}

type postTweetOutput struct {
    TweetID string `json:"tweet_id"`
    // maybe other info off the `domain.Tweet` type which could be useful
    // to client state management, assuming the API is consumed by an SPA, in React for example. 

It may look similar to domain.Domain, but notice the difference in the input parameters. The Handler sits at the edge between the client and the server and is responsible of:

• the conversion of the data from and to a format understandable by the user
• converting errors to a format understandable by and useful to the user (this is in a way a subset of the first rule)
  What Handler.PostTweet does in our case is the following:
• read the data from the ResponseWriter and parse it as JSON, using the unexported postTweetInput helper. Sends a 400 response if the JSON is invalid
• pass that data to domain.PostTweet and handle the error: return 400 for validation errors, 404 for user not found, 500 for database errors
• create a postTweetOutput from the resulted domain.Tweet, serialize it as JSON and write it to the client.

Note: to distinguish between the error types returned by the domain layer you may have a domain.Error type and a field on it of type domain.ErrorKind, which could be an enum like so:

package domain

type ErrorKind int

const (
    ErrorKindInvalidInput ErrorKind = iota
    ErrorKindNotFound
    ErrorKindInternal
)

type Error struct {
    Kind ErrorKind
    // Cause is the original error. May be from validation, from the database etc. 
    Cause error
    // other useful fields maybe
}

// Error implements the error interface.
func (e *Error) Error() string
// Unwrap returns e.Cause. It makes this type compatible with errors.As, errors.Unwrap and others.
func (e *Error) Unwrap() error

This would be our route handler. In a way you can see how in this case it's just a soft wrapper around the domain layer. If your application is simple and this is too much boilerplate you could omit it and just have the data layer – that still gives you most of the benefits. In the real world, for example, some business logic operations are reused, and by not having the layer code would be duplicated. The separation is also worth it because for example authentication/authorization may be involved and you wouldn't want to execute any business logic for unauthenticated users; by not mixing the two this can be ensured not to happen, as one would authorize first and then go to the domain layer (in code there could be an Authorizer struct similar to the Domain one, which could also be an interface, to allow swapping authentication providers down the line, e.g. moving from Amazon Cognito to Google auth).

Finally, how are all these glued up? Here's how the main code could look like:

func main() {
    gormDB, err := /* init the database */
    // handle the error

    sseHandler := /* init the SSE handler */

    repo := gormdata.New(gormDB) // assume the constructors exists in their respective packages
    dom := domain.New(repo, sseHandler)
    handler := handler.New(dom)

    mux := http.NewServeMux()
    // Here handler.PostTweet could be wrapped in middleware, if necessary
    mux.HandleFunc("POST /post", handler.PostTweet)

    s := &http.Server{
        Address: "0.0.0.0:8080",
        Handler: mux, // Could also be wrapped in middleware: a panic handler, CORS, logging etc.
        // other fields
    }
    s.RegisterOnShutdown(func() {
        // the SSE close message code
    })

    // maybe add graceful shutdown

    err := s.ListenAndServe()
    // handle the error
}

Note: graceful shutdown is also implemented in the complex example. Take inspiration from that main function for your server startup code.

This would be the general outline.

How does this long-ass rushed incomplete high-level three tier architecture blog post relate to your AfterUpdate issue, though? The data/domain separation may have sparked an idea: dispatching status change events is the responsibility of the domain layer, not the data layer. I don't know what a Node is, but what I am pretty sure of is that its state updates come from some user operations. You should know which operations change the Node state – therein should the events be dispatched, not in the data layer (in your case the GORM hook). Regardless if you do consider these layers as separate entities in your code or not, they still exist – even if you don't explicitly write your code as if they're separated, I'd still try to think of this separation and honor it conceptually and implicitly. It will make your code cleaner, more straight forward in the long run and help you better shape its architecture and think the business requirements properly through.

Final notes

Hopefully this helped in giving you some direction on go-sse and Go in general! Let me know if you need any clarifications.

[irreal]

I know this is what the emoji reactions are for, but I see the OP hasn't replied in some time and I just had to write a thank you note!

As someone searching for a good way to fit SSE into my app architecture this was an amazing read full of great detail and useful links.

Thank you so much for not only creating the library but also being so dedicated and helpful in all the issues I've browsed so far. Kudos to you, great job! Just wanted you to know you are very much appreciated. ❤️

[tmaxmax]

Thank you too for the very kind words! Glad to find out this was helpful to you.