add server-sent events handler

cmd/server/main.go

@@ -207,6 +207,7 @@ func runroomsrv() error {
 	roomsrv, err := mksrv.New(
 		db.Members,
 		db.Aliases,
+		db.AuthWithSSB,
 		httpsDomain,
 		opts...)
 	if err != nil {

go.mod

@@ -32,6 +32,7 @@ require (
 	go.mindeco.de v1.9.0
 	go.mindeco.de/ssb-refs v0.1.1-0.20210108133850-cf1f44fea870
 	golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad
+	golang.org/x/net v0.0.0-20191116160921-f9c825593386 // indirect
 	golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9
 	golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c // indirect
 	golang.org/x/text v0.3.5

go.sum

@@ -456,8 +456,6 @@ go.cryptoscope.co/margaret v0.0.5/go.mod h1:W+Q6lvzHIrF8+Yt3dItHHsx2R9/Xvj/NkJGM
 go.cryptoscope.co/margaret v0.0.8/go.mod h1:VbP0bqavqW5osTmdNvgukrqtmqvZC5sXyPSBUW5rzv8=
 go.cryptoscope.co/margaret v0.0.12-0.20190912103626-34323ad497f4 h1:gLSldWRujtUOfdnpA1XKD71xcCp3Wz1URMnT6xpUPV4=
 go.cryptoscope.co/margaret v0.0.12-0.20190912103626-34323ad497f4/go.mod h1:3rt+RmZTFZEgfvFxz0ZPDBIWtLJOouWtzV6YbBl6sek=
-go.cryptoscope.co/muxrpc/v2 v2.0.0-20210202162901-fe642d405dc6 h1:p135TwijE3DbmklGygc7++MMRRVlujmjqed8kEOmwLs=
-go.cryptoscope.co/muxrpc/v2 v2.0.0-20210202162901-fe642d405dc6/go.mod h1:MgaeojIkWY3lLuoNw1mlMT3b3jiZwOj/fgsoGZp/VNA=
 go.cryptoscope.co/muxrpc/v2 v2.0.0-beta.1.0.20210308090127-5f1f5f9cbb59 h1:Gv5pKkvHYJNc12uRZ/jMCsR17G7v6oFLLCrGAUVxhvo=
 go.cryptoscope.co/muxrpc/v2 v2.0.0-beta.1.0.20210308090127-5f1f5f9cbb59/go.mod h1:MgaeojIkWY3lLuoNw1mlMT3b3jiZwOj/fgsoGZp/VNA=
 go.cryptoscope.co/netwrap v0.1.0/go.mod h1:7zcYswCa4CT+ct54e9uH9+IIbYYETEMHKDNpzl8Ukew=
@@ -511,8 +509,9 @@ golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR
 golang.org/x/net v0.0.0-20190607181551-461777fb6f67/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7 h1:fHDIZ2oxGnUZRN6WgWFCbYBjH9uqVPRCUVUDhs0wnbA=
 golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20191116160921-f9c825593386 h1:ktbWvQrW08Txdxno1PiDpSxPXG6ndGsfnJjRRtkM0LQ=
+golang.org/x/net v0.0.0-20191116160921-f9c825593386/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=

roomdb/sqlite/new_test.go

@@ -8,9 +8,10 @@ import (
 	"testing"
 
 	_ "github.com/mattn/go-sqlite3"
+	"github.com/stretchr/testify/require"
+
 	"github.com/ssb-ngi-pointer/go-ssb-room/internal/repo"
 	"github.com/ssb-ngi-pointer/go-ssb-room/roomdb"
-	"github.com/stretchr/testify/require"
 	refs "go.mindeco.de/ssb-refs"
 )
 

roomsrv/init_handlers.go

@@ -34,6 +34,10 @@ func (s *Server) initHandlers(aliasDB roomdb.AliasesService) {
 
 	siwssbHandler := signinwithssb.New(
 		kitlog.With(s.logger, "unit", "auth-with-ssb"),
+		s.Whoami(),
+		s.authWithSSB,
+		s.Members,
+		s.domain,
 	)
 
 	// register muxrpc commands

roomsrv/server.go

@@ -66,6 +66,8 @@ type Server struct {
 
 	Members roomdb.MembersService
 	Aliases roomdb.AliasesService
+
+	authWithSSB roomdb.AuthWithSSBService
 }
 
 func (s Server) Whoami() refs.FeedRef {
@@ -75,6 +77,7 @@ func (s Server) Whoami() refs.FeedRef {
 func New(
 	membersdb roomdb.MembersService,
 	aliasdb roomdb.AliasesService,
+	awsdb roomdb.AuthWithSSBService,
 	domainName string,
 	opts ...Option,
 ) (*Server, error) {
@@ -84,6 +87,8 @@ func New(
 	s.Members = membersdb
 	s.Aliases = aliasdb
 
+	s.authWithSSB = awsdb
+
 	s.domain = domainName
 
 	for i, opt := range opts {

web/assets/events-demo.js

@@ -0,0 +1,13 @@
+let streamName = document.querySelector("#stream-name").attributes.stream.value
+
+var evtSource = new EventSource(`/sse/events?stream=${streamName}`);
+
+var eventList = document.querySelector('#event-list');
+
+evtSource.addEventListener("testing", (e) => {
+//   console.log(e)
+
+  var newElement = document.createElement("li");
+  newElement.textContent = `(${e.lastEventId}) message: ${e.data}`;
+  eventList.prepend(newElement);
+})

web/handlers/auth/handler.go

@@ -15,7 +15,9 @@ import (
 
 var HTMLTemplates = []string{
 	"auth/fallback_sign_in.tmpl",
+
 	"auth/withssb_sign_in.tmpl",
+	"auth/withssb_server_start.tmpl",
 }
 
 func NewFallbackPasswordHandler(

web/handlers/auth/withssb.go

@@ -5,17 +5,23 @@ package auth
 import (
 	"context"
 	"encoding/base64"
+	"encoding/binary"
 	"encoding/gob"
 	"fmt"
 	"net/http"
 	"time"
 
+	kitlog "github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log/level"
 	"github.com/gorilla/mux"
 	"github.com/gorilla/sessions"
+
 	"go.cryptoscope.co/muxrpc/v2"
 	"go.mindeco.de/http/render"
+	"go.mindeco.de/logging"
 
 	"github.com/ssb-ngi-pointer/go-ssb-room/internal/network"
+	"github.com/ssb-ngi-pointer/go-ssb-room/internal/randutil"
 	"github.com/ssb-ngi-pointer/go-ssb-room/internal/signinwithssb"
 	"github.com/ssb-ngi-pointer/go-ssb-room/roomdb"
 	weberrors "github.com/ssb-ngi-pointer/go-ssb-room/web/errors"
@@ -58,6 +64,10 @@ func NewWithSSBHandler(
 
 	m.Get(router.AuthWithSSBSignIn).HandlerFunc(r.HTML("auth/withssb_sign_in.tmpl", ssb.login))
 
+	m.HandleFunc("/sse/login/{sc}", r.HTML("auth/withssb_server_start.tmpl", ssb.startWithServer))
+
+	m.HandleFunc("/sse/events", ssb.eventSource)
+
 	return &ssb
 }
 
@@ -253,3 +263,126 @@ func (h WithSSBHandler) Logout(w http.ResponseWriter, r *http.Request) {
 	http.Redirect(w, r, "/", http.StatusSeeOther)
 	return
 }
+
+// server-sent-events stuff
+
+func (h WithSSBHandler) startWithServer(w http.ResponseWriter, req *http.Request) (interface{}, error) {
+	logger := logging.FromContext(req.Context())
+
+	streamName := randutil.String(20)
+	// h.events.CreateStream(streamName)
+
+	logger = level.Debug(logger)
+	logger = kitlog.With(logger, "event", streamName)
+
+	logger.Log("event", "started stream")
+
+	// tick := time.NewTicker(5 * time.Second)
+	// go func() {
+
+	// 	var (
+	// 		evtBuf = make([]byte, 4)
+	// 		evtID  = uint32(0)
+	// 	)
+	// 	for range tick.C {
+	// 		binary.BigEndian.PutUint32(evtBuf, evtID)
+	// 		evtID++
+	// 		h.events.Publish(streamName, &sse.Event{
+	// 			ID:    evtBuf,
+	// 			Data:  []byte(fmt.Sprintf("boring: %d", evtID)),
+	// 			Event: []byte("testing"),
+	// 		})
+	// 		logger.Log("event", "sent", "id", evtID)
+	// 	}
+	// }()
+
+	// go func() {
+	// 	time.Sleep(1 * time.Minute)
+	// 	tick.Stop()
+	// 	logger.Log("event", "stopped")
+	// }()
+
+	return struct {
+		StreamName string
+	}{streamName}, nil
+}
+
+type event struct {
+	ID, Data, Event []byte
+}
+
+func (h WithSSBHandler) eventSource(w http.ResponseWriter, r *http.Request) {
+	flusher, err := w.(http.Flusher)
+	if !err {
+		http.Error(w, "Streaming unsupported!", http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	w.Header().Set("Transfer-Encoding", "chunked")
+
+	// Get the StreamID from the URL
+	streamID := r.URL.Query().Get("stream")
+	if streamID == "" {
+		http.Error(w, "Please specify a stream!", http.StatusInternalServerError)
+		return
+	}
+	logger := logging.FromContext(r.Context())
+	logger = level.Debug(logger)
+	logger = kitlog.With(logger, "stream", streamID)
+
+	logger.Log("event", "stream opened")
+
+	// TODO: load map with channel
+
+	// tick := time.NewTicker(time.Second / 4)
+	tick := time.NewTicker(time.Second)
+	notify := w.(http.CloseNotifier).CloseNotify()
+	go func() {
+		<-notify
+		tick.Stop()
+		logger.Log("event", "request closed")
+	}()
+
+	go func() {
+		time.Sleep(5 * time.Minute)
+		tick.Stop()
+		logger.Log("event", "stopped")
+	}()
+
+	start := time.Now()
+	flusher.Flush()
+
+	// Push events to client
+	var (
+		evtBuf = make([]byte, 4)
+		evtID  = uint32(0)
+		evt    event
+	)
+
+	for range tick.C {
+		binary.BigEndian.PutUint32(evtBuf, evtID)
+		evtID++
+
+		evt = event{
+			ID:    []byte(fmt.Sprintf("%d", evtID)),
+			Data:  []byte(fmt.Sprintf("age: %s", time.Since(start))),
+			Event: []byte("testing"),
+		}
+
+		fmt.Fprintf(w, "id: %s\n", evt.ID)
+		fmt.Fprintf(w, "data: %s\n", evt.Data)
+		if len(evt.Event) > 0 {
+			fmt.Fprintf(w, "event: %s\n", evt.Event)
+		}
+		// if len(evt.Retry) > 0 {
+		// 	fmt.Fprintf(w, "retry: %s\n", evt.Retry)
+		// }
+		fmt.Fprint(w, "\n")
+		flusher.Flush()
+
+		logger.Log("event", "sent", "id", evtID)
+	}
+}

web/templates/auth/withssb_server_start.tmpl

@@ -0,0 +1,15 @@
+{{ define "title" }}{{i18n "AuthWithSSBTitle"}}{{ end }}
+{{ define "content" }}
+      <div id="page-header">
+        <h1 id="welcome" class="text-lg">{{i18n "AuthWithSSBWelcome"}}</h1>
+      </div>
+      <div>
+        <h3>Template Data</h3>
+        <pre>{{.}}</pre>
+
+        <h3>Server events</h3>
+        <ul id="event-list"></ul>
+      </div>
+      <div id="stream-name" stream="{{.StreamName}}"></div>
+      <script src="/assets/events-demo.js"></script>
+{{end}}