ui.go

package rangedbui

import (
	"context"
	"embed"
	"encoding/json"
	"errors"
	"fmt"
	"html/template"
	"io/fs"
	"log"
	"net/http"
	"time"

	"github.com/gorilla/handlers"
	"github.com/gorilla/mux"
	"github.com/gorilla/websocket"

	"github.com/inklabs/rangedb"
	"github.com/inklabs/rangedb/pkg/paging"
	"github.com/inklabs/rangedb/pkg/projection"
)

//go:embed static
var StaticAssets embed.FS

//go:embed templates
var Templates embed.FS

const (
	subscriberRecordBuffSize = 20
	defaultHost              = "0.0.0.0:8080"
)

type webUI struct {
	handler            http.Handler
	aggregateTypeStats *projection.AggregateTypeStats
	store              rangedb.Store
	templateFS         fs.FS
	upgrader           *websocket.Upgrader
	host               string
}

// Option defines functional option parameters for webUI.
type Option func(*webUI)

// WithTemplateFS is a functional option to inject a template filesystem
func WithTemplateFS(f fs.FS) Option {
	return func(webUI *webUI) {
		webUI.templateFS = f
	}
}

// WithHost is a functional option to inject a tcp4 host.
func WithHost(host string) Option {
	return func(app *webUI) {
		app.host = host
	}
}

// New constructs a webUI web application.
func New(
	aggregateTypeStats *projection.AggregateTypeStats,
	store rangedb.Store,
	options ...Option,
) *webUI {
	webUI := &webUI{
		aggregateTypeStats: aggregateTypeStats,
		store:              store,
		templateFS:         Templates,
		upgrader: &websocket.Upgrader{
			ReadBufferSize:  1024,
			WriteBufferSize: 1024,
			CheckOrigin: func(r *http.Request) bool {
				return true
			},
		},
		host: defaultHost,
	}

	for _, option := range options {
		option(webUI)
	}

	webUI.initRoutes()

	return webUI
}

func (a *webUI) initRoutes() {
	const streamName = "{streamName}"
	const aggregateType = "{aggregateType:[a-zA-Z-]+}"
	router := mux.NewRouter().StrictSlash(true)

	main := router.PathPrefix("/").Subrouter()
	main.HandleFunc("/", a.index)
	main.HandleFunc("/aggregate-types", a.aggregateTypes)
	main.HandleFunc("/aggregate-types/live", a.aggregateTypesLive)
	main.HandleFunc("/a/"+aggregateType, a.aggregateType)
	main.HandleFunc("/a/"+aggregateType+"/live", a.aggregateTypeLive)
	main.HandleFunc("/s/"+streamName, a.stream)
	main.PathPrefix("/static/").Handler(http.FileServer(http.FS(StaticAssets)))
	main.Use(handlers.CompressHandler)

	websocketRouter := router.PathPrefix("/live").Subrouter()
	websocketRouter.HandleFunc("/a/"+aggregateType, a.realtimeEventsByAggregateType)
	websocketRouter.HandleFunc("/aggregate-types", a.realtimeAggregateTypes)

	a.handler = router
}

func (a *webUI) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	a.handler.ServeHTTP(w, r)
}

func (a *webUI) index(w http.ResponseWriter, r *http.Request) {
	http.Redirect(w, r, "/aggregate-types", http.StatusFound)
}

type aggregateTypesTemplateVars struct {
	AggregateTypes []AggregateTypeInfo
	TotalEvents    uint64
	UIHost         string
}

type liveAggregateTypesTemplateVars struct {
	AggregateTypes []AggregateTypeInfo
	TotalEvents    uint64
}

func (a *webUI) aggregateTypes(w http.ResponseWriter, _ *http.Request) {
	var aggregateTypes []AggregateTypeInfo

	for _, aggregateType := range a.aggregateTypeStats.SortedAggregateTypes() {
		aggregateTypes = append(aggregateTypes, AggregateTypeInfo{
			Name:        aggregateType,
			TotalEvents: a.aggregateTypeStats.TotalEventsByAggregateType(aggregateType),
		})
	}

	a.renderWithValues(w, "aggregate-types.gohtml", aggregateTypesTemplateVars{
		AggregateTypes: aggregateTypes,
		TotalEvents:    a.aggregateTypeStats.TotalEvents(),
	})
}

func (a *webUI) aggregateTypesLive(w http.ResponseWriter, _ *http.Request) {
	var aggregateTypes []AggregateTypeInfo

	for _, aggregateType := range a.aggregateTypeStats.SortedAggregateTypes() {
		aggregateTypes = append(aggregateTypes, AggregateTypeInfo{
			Name:        aggregateType,
			TotalEvents: a.aggregateTypeStats.TotalEventsByAggregateType(aggregateType),
		})
	}

	a.renderWithValues(w, "aggregate-types-live.gohtml", aggregateTypesTemplateVars{
		AggregateTypes: aggregateTypes,
		TotalEvents:    a.aggregateTypeStats.TotalEvents(),
		UIHost:         a.host,
	})
}

type aggregateTypeTemplateVars struct {
	AggregateTypeInfo AggregateTypeInfo
	PaginationLinks   paging.Links
	Records           []*rangedb.Record
	UIHost            string
}

func (a *webUI) aggregateType(w http.ResponseWriter, r *http.Request) {
	aggregateTypeName := mux.Vars(r)["aggregateType"]
	pagination := paging.NewPaginationFromQuery(r.URL.Query())

	globalSequenceNumber := pagination.CurrentIndex
	records := rangedb.ReadNRecords(
		pagination.ItemsPerPage+1,
		func() (rangedb.RecordIterator, context.CancelFunc) {
			ctx, done := context.WithCancel(r.Context())
			return a.store.EventsByAggregateTypes(ctx, globalSequenceNumber, aggregateTypeName), done
		},
	)

	nextIndex := pagination.CurrentIndex
	if len(records) > int(pagination.ItemsPerPage) {
		lastRecord := records[len(records)-1]
		nextIndex = lastRecord.GlobalSequenceNumber

		records = records[:len(records)-1]
	}

	baseURI := fmt.Sprintf("/a/%s", aggregateTypeName)
	totalRecords := a.aggregateTypeStats.TotalEventsByAggregateType(aggregateTypeName)

	a.renderWithValues(w, "aggregate-type.gohtml", aggregateTypeTemplateVars{
		AggregateTypeInfo: AggregateTypeInfo{
			Name:        aggregateTypeName,
			TotalEvents: totalRecords,
		},
		PaginationLinks: pagination.Links(baseURI, nextIndex),
		Records:         records,
	})
}

type streamTemplateVars struct {
	PaginationLinks paging.Links
	Records         []*rangedb.Record
	StreamInfo      StreamInfo
}

func (a *webUI) stream(w http.ResponseWriter, r *http.Request) {
	streamName := mux.Vars(r)["streamName"]
	pagination := paging.NewPaginationFromQuery(r.URL.Query())

	streamSequenceNumber := pagination.CurrentIndex
	records := rangedb.ReadNRecords(
		pagination.ItemsPerPage+1,
		func() (rangedb.RecordIterator, context.CancelFunc) {
			ctx, done := context.WithCancel(r.Context())
			return a.store.EventsByStream(ctx, streamSequenceNumber, streamName), done
		},
	)

	nextIndex := pagination.CurrentIndex
	if len(records) > int(pagination.ItemsPerPage) {
		lastRecord := records[len(records)-1]
		nextIndex = lastRecord.StreamSequenceNumber

		records = records[:len(records)-1]
	}

	baseURI := fmt.Sprintf("/s/%s", streamName)
	totalRecords, err := a.store.TotalEventsInStream(r.Context(), streamName)
	if err != nil {
		if errors.Is(err, context.Canceled) {
			http.Error(w, "Request Timeout", http.StatusRequestTimeout)
			return
		}

		http.Error(w, "500 Internal Server Error", http.StatusInternalServerError)
		return
	}

	a.renderWithValues(w, "stream.gohtml", streamTemplateVars{
		PaginationLinks: pagination.Links(baseURI, nextIndex),
		Records:         records,
		StreamInfo: StreamInfo{
			Name:        streamName,
			TotalEvents: totalRecords,
		},
	})
}

func (a *webUI) aggregateTypeLive(w http.ResponseWriter, r *http.Request) {
	aggregateTypeName := mux.Vars(r)["aggregateType"]

	totalRecords := a.aggregateTypeStats.TotalEventsByAggregateType(aggregateTypeName)

	a.renderWithValues(w, "aggregate-type-live.gohtml", aggregateTypeTemplateVars{
		AggregateTypeInfo: AggregateTypeInfo{
			Name:        aggregateTypeName,
			TotalEvents: totalRecords,
		},
		UIHost: a.host,
	})
}

func (a *webUI) renderWithValues(w http.ResponseWriter, tpl string, data interface{}) {
	w.Header().Set("Content-Type", "text/html; charset=utf-8")

	err := a.RenderTemplate(w, tpl, data)
	if err != nil {
		log.Printf("unable to render template %s: %v", tpl, err)
		http.Error(w, "500 Internal Server Error", http.StatusInternalServerError)
		return
	}
}

func (a *webUI) RenderTemplate(w http.ResponseWriter, tpl string, data interface{}) error {
	tmpl, err := template.New(tpl).Funcs(FuncMap).ParseFS(a.templateFS, "templates/layout/*.gohtml", "templates/*"+tpl)
	if err != nil {
		return fmt.Errorf("unable to parse template: %w", err)
	}

	return tmpl.Execute(w, data)
}

type RecordEnvelope struct {
	Record      rangedb.Record
	TotalEvents uint64
}

func (a *webUI) realtimeEventsByAggregateType(w http.ResponseWriter, r *http.Request) {
	aggregateTypeName := mux.Vars(r)["aggregateType"]

	conn, err := a.upgrader.Upgrade(w, r, nil)
	if err != nil {
		http.Error(w, "unable to upgrade websocket connection", http.StatusBadRequest)
		return
	}
	defer conn.Close()

	// keepAlive(conn, 1*time.Minute)

	latestGlobalSequenceNumber := a.aggregateTypeStats.LatestGlobalSequenceNumber()
	totalRecords := a.aggregateTypeStats.TotalEventsByAggregateType(aggregateTypeName)
	ctx, cancel := context.WithCancel(r.Context())

	recordSubscriber := rangedb.RecordSubscriberFunc(func(record *rangedb.Record) {
		totalRecords++

		envelope := RecordEnvelope{
			Record:      *record,
			TotalEvents: totalRecords,
		}

		message, err := json.Marshal(envelope)
		if err != nil {
			log.Printf("unable to marshal record: %v", err)
			return
		}

		writeErr := conn.WriteMessage(websocket.TextMessage, message)
		if writeErr != nil {
			log.Printf("unable to write to ws client: %v", writeErr)
			cancel()
		}
	})

	subscription := a.store.AggregateTypesSubscription(ctx, subscriberRecordBuffSize, recordSubscriber, aggregateTypeName)
	defer subscription.Stop()
	err = subscription.StartFrom(latestGlobalSequenceNumber)
	if err != nil {
		log.Printf("unable to start subscription: %v", err)
		return
	}

	_, _, _ = conn.ReadMessage()
}

func (a *webUI) realtimeAggregateTypes(w http.ResponseWriter, r *http.Request) {
	conn, err := a.upgrader.Upgrade(w, r, nil)
	if err != nil {
		http.Error(w, "unable to upgrade websocket connection", http.StatusBadRequest)
		return
	}
	defer conn.Close()

	// keepAlive(conn, 1*time.Minute)

	latestGlobalSequenceNumber := uint64(0)

	for {
		time.Sleep(100 * time.Millisecond)

		if a.aggregateTypeStats.LatestGlobalSequenceNumber() <= latestGlobalSequenceNumber {
			continue
		}

		latestGlobalSequenceNumber = a.aggregateTypeStats.LatestGlobalSequenceNumber()

		var aggregateTypes []AggregateTypeInfo
		for _, aggregateType := range a.aggregateTypeStats.SortedAggregateTypes() {
			aggregateTypes = append(aggregateTypes, AggregateTypeInfo{
				Name:        aggregateType,
				TotalEvents: a.aggregateTypeStats.TotalEventsByAggregateType(aggregateType),
			})
		}

		envelope := liveAggregateTypesTemplateVars{
			AggregateTypes: aggregateTypes,
			TotalEvents:    a.aggregateTypeStats.TotalEvents(),
		}

		message, err := json.Marshal(envelope)
		if err != nil {
			log.Printf("unable to marshal record: %v", err)
			return
		}

		writeErr := conn.WriteMessage(websocket.TextMessage, message)
		if writeErr != nil {
			log.Printf("unable to write to ws client: %v", writeErr)
			return
		}
	}
}

func keepAlive(c *websocket.Conn, timeout time.Duration) {
	lastResponse := time.Now()
	c.SetPongHandler(func(_ string) error {
		lastResponse = time.Now()
		return nil
	})

	go func() {
		for {
			err := c.WriteMessage(websocket.PingMessage, []byte("keepalive"))
			if err != nil {
				return
			}
			time.Sleep(timeout / 2)
			if time.Since(lastResponse) > timeout {
				_ = c.Close()
				return
			}
		}
	}()
}

// AggregateTypeInfo contains the aggregate type data available to templates.
type AggregateTypeInfo struct {
	Name        string
	TotalEvents uint64
}

// StreamInfo contains the stream data available to templates.
type StreamInfo struct {
	Name        string
	TotalEvents uint64
}