sse support
This commit is contained in:
@@ -23,22 +23,21 @@ func Routes(
|
||||
r *gin.RouterGroup,
|
||||
logger *logging.Logger,
|
||||
accts *accounts.Store,
|
||||
authMiddleware *middleware.Auth,
|
||||
pub *middleware.UpdateNotificationPublisher,
|
||||
) {
|
||||
as := &accountSubrouter{
|
||||
log: logger,
|
||||
accts: accts,
|
||||
}
|
||||
|
||||
withAuth := func(fn response.HandlerFunc) gin.HandlerFunc {
|
||||
return response.Handler(authMiddleware.AuthenticateAndAddIdentity(fn))
|
||||
}
|
||||
syncGroups := r.Group("/:acctID/inventory/sync-groups")
|
||||
syncGroups.POST("", response.Handler(as.saveNewSyncGroup))
|
||||
|
||||
r.POST("/:acctID/inventory/sync-groups/draft/listings", withAuth(as.createSyncGroupListingDraft))
|
||||
r.PUT("/:acctID/inventory/sync-groups/draft/listings/:orderIndex/shop", withAuth(as.setShopInSyncGroupListingDraft))
|
||||
r.PUT("/:acctID/inventory/sync-groups/draft/listings/:orderIndex/listing", withAuth(as.setListingInSyncGroupListingDraft))
|
||||
r.DELETE("/:acctID/inventory/sync-groups/draft/listings/:orderIndex", withAuth(as.deleteSyncGroupListingDraft))
|
||||
r.POST("/:acctID/inventory/sync-groups", withAuth(as.saveNewSyncGroup))
|
||||
draftListings := syncGroups.Group("/draft/listings", pub.Publish("/:acctID/inventory/sync-groups/draft/listings"))
|
||||
draftListings.POST("", response.Handler(as.createSyncGroupListingDraft))
|
||||
draftListings.PUT("/:orderIndex/shop", response.Handler(as.setShopInSyncGroupListingDraft))
|
||||
draftListings.PUT("/:orderIndex/listing", response.Handler(as.setListingInSyncGroupListingDraft))
|
||||
draftListings.DELETE("/:orderIndex", response.Handler(as.deleteSyncGroupListingDraft))
|
||||
}
|
||||
|
||||
// POST /api/accounts
|
||||
|
||||
@@ -0,0 +1,97 @@
|
||||
package sse
|
||||
|
||||
import (
|
||||
"context"
|
||||
"ruben/inventory2/internal/logging"
|
||||
"ruben/inventory2/internal/server/middleware"
|
||||
"ruben/inventory2/internal/server/response"
|
||||
"ruben/inventory2/internal/server/sse"
|
||||
"sync"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
type (
|
||||
sseRouter struct {
|
||||
log *logging.Logger
|
||||
sse *sse.Queue
|
||||
|
||||
users map[string]context.CancelFunc
|
||||
lock sync.Mutex
|
||||
}
|
||||
)
|
||||
|
||||
func Routes(
|
||||
r gin.IRouter,
|
||||
logger *logging.Logger,
|
||||
sq *sse.Queue,
|
||||
auth *middleware.Auth,
|
||||
) {
|
||||
s := &sseRouter{
|
||||
log: logger,
|
||||
sse: sq,
|
||||
users: make(map[string]context.CancelFunc),
|
||||
}
|
||||
|
||||
r.GET("/", auth.AuthenticateAndAddIdentityGin(), response.Handler(s.serveEvents))
|
||||
}
|
||||
|
||||
func (r *sseRouter) serveEvents(c *gin.Context) (response.Response, error) {
|
||||
acct := middleware.GetIdentity(c).Account
|
||||
acctID := acct.AccountID
|
||||
userID := acct.UserID
|
||||
email := acct.Email
|
||||
|
||||
log := r.log.WithGroup("serveEvents").With(
|
||||
"accountID", acctID,
|
||||
"userID", userID,
|
||||
"email", email,
|
||||
)
|
||||
log.Info("user connected to sse queue")
|
||||
|
||||
ctx := r.closeExistingConnectionsForUserAndStoreCancelFuncForUser(c, userID)
|
||||
|
||||
w := c.Writer
|
||||
|
||||
hdr := w.Header()
|
||||
hdr.Set("Access-Control-Allow-Origin", "*")
|
||||
hdr.Set("Access-Control-Expose-Headers", "Content-Type")
|
||||
hdr.Set("Content-Type", "text/event-stream")
|
||||
hdr.Set("Connection", "keep-alive")
|
||||
hdr.Set("Cache-Control", "no-cache")
|
||||
w.Flush()
|
||||
|
||||
err := r.sse.Listen(ctx, func(ctx context.Context, e *sse.Event) error {
|
||||
log.Debugf("sending event of type %s", e.Type)
|
||||
|
||||
e.Write(w)
|
||||
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
log.Errorf("no longer connected to sse queue due to error: %v", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// send a 'close' message so the front end doesn't try to reconnect
|
||||
(&sse.Event{Type: "close"}).Write(w)
|
||||
|
||||
log.Info("user disconnecting sse queue")
|
||||
return response.Status(200), nil
|
||||
}
|
||||
|
||||
func (r *sseRouter) closeExistingConnectionsForUserAndStoreCancelFuncForUser(ctx context.Context, userID string) context.Context {
|
||||
r.lock.Lock()
|
||||
defer r.lock.Unlock()
|
||||
|
||||
// close existing connection
|
||||
if closeConn, ok := r.users[userID]; ok {
|
||||
closeConn()
|
||||
delete(r.users, userID)
|
||||
}
|
||||
|
||||
// store reference to cancel func
|
||||
ctx, r.users[userID] = context.WithCancel(ctx)
|
||||
|
||||
return ctx
|
||||
}
|
||||
@@ -111,13 +111,17 @@ func (a *Auth) AuthenticateAndAddIdentity(f response.HandlerFunc, assertions ...
|
||||
return func(c *gin.Context) (response.Response, error) {
|
||||
c, res, err := a.AuthenticateAndAddIdentityToRequest(c, assertions...)
|
||||
if res != nil || err != nil {
|
||||
return res, nil
|
||||
return res, err
|
||||
}
|
||||
|
||||
return f(c)
|
||||
}
|
||||
}
|
||||
|
||||
func (a *Auth) AuthenticateAndAddIdentityGin(assertions ...AuthorizationAssertions) gin.HandlerFunc {
|
||||
return a.AuthenticateAndAddIdentityToRequestGin(assertions...)
|
||||
}
|
||||
|
||||
func (a *Auth) AuthenticateAndAddIdentityToRequest(c *gin.Context, assertions ...AuthorizationAssertions) (*gin.Context, response.Response, error) {
|
||||
r := c.Request
|
||||
ck, err := r.Cookie("access_token")
|
||||
@@ -188,15 +192,129 @@ func (a *Auth) AuthenticateAndAddIdentityToRequest(c *gin.Context, assertions ..
|
||||
return c, nil, nil
|
||||
}
|
||||
|
||||
func (a *Auth) AuthenticateAndAddIdentityToRequestGin(assertions ...AuthorizationAssertions) gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
r := c.Request
|
||||
ck, err := r.Cookie("access_token")
|
||||
if err != nil {
|
||||
response.Write(c, response.TemporaryRedirect("/").
|
||||
JSON("no access_token cookie provided"))
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
|
||||
ctx := r.Context()
|
||||
|
||||
accessToken := ck.Value
|
||||
|
||||
claims, expiration, err := a.auth.GetAccessTokenClaimsAndExpiration(ctx, accessToken)
|
||||
if err != nil {
|
||||
if errors.Is(err, consts.ErrNotFound) {
|
||||
u, err := a.newLoginURL(ctx, a.auth, r.URL.String())
|
||||
if err != nil {
|
||||
response.WriteError(c, response.Errorf("failed to generate login url: %w", err))
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
|
||||
response.Write(c, response.TemporaryRedirect(u))
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
|
||||
response.WriteError(c, response.Errorf("failed to authenticate: %w", err))
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
|
||||
// refresh tokens, when the access token is "old enough"
|
||||
|
||||
// id token lifetime is 48 hours, allowing a person to use the app everyday comfortably, with wiggle room, without having to log in.
|
||||
const idTokenLifetime = 48 * time.Hour
|
||||
if refreshFloor := expiration.Add(-(idTokenLifetime / 4)); refreshFloor.Before(now) {
|
||||
accessToken, expiration, err = a.auth.RefreshAccessToken(ctx, accessToken)
|
||||
if err != nil {
|
||||
a.log.Warn("failed to refresh access token", "error", err)
|
||||
response.Write(c, response.TemporaryRedirect("/").
|
||||
Body(io.NopCloser(bytes.NewBuffer([]byte(fmt.Sprintf("failed to refresh access token: %v", err))))).
|
||||
Cookie(cookies.Expired("access_token")))
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
|
||||
// 'redirect' to same url, to set the new access_token cookie
|
||||
response.Write(c, response.TemporaryRedirect(r.URL.String()).
|
||||
Cookie(cookies.AccessToken(accessToken, expiration)))
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
|
||||
// add identity info to request context
|
||||
|
||||
user, acct, err := a.accts.GetUserAndAccountByAccessToken(ctx, accessToken)
|
||||
if err != nil {
|
||||
response.WriteError(c, response.Errorf("failed to authorize: %w", err))
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
|
||||
for _, as := range assertions {
|
||||
res, err := as(c)
|
||||
if err != nil {
|
||||
response.WriteError(c, err)
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
if res != nil {
|
||||
response.Write(c, res)
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
id := Identity{
|
||||
AccessToken: accessToken,
|
||||
Claims: claims,
|
||||
User: user,
|
||||
Account: acct,
|
||||
}
|
||||
|
||||
c.Set(identityKeyString, id)
|
||||
c.Request = r.WithContext(SetIdentity(ctx, id))
|
||||
|
||||
c.Next()
|
||||
}
|
||||
}
|
||||
|
||||
type identityKey struct{}
|
||||
|
||||
// stores identity in request context
|
||||
const identityKeyString = "identity"
|
||||
|
||||
// stores identity in context
|
||||
func SetIdentity(ctx context.Context, id Identity) context.Context {
|
||||
return context.WithValue(ctx, identityKey{}, id)
|
||||
}
|
||||
|
||||
// get identity from request context
|
||||
// get identity from context
|
||||
func GetIdentity(ctx context.Context) Identity {
|
||||
id, _ := ctx.Value(identityKey{}).(Identity)
|
||||
id, ok := ctx.Value(identityKey{}).(Identity)
|
||||
if ok {
|
||||
return id
|
||||
}
|
||||
|
||||
c, ok := ctx.(*gin.Context)
|
||||
if !ok {
|
||||
return Identity{}
|
||||
}
|
||||
|
||||
v, ok := c.Get(identityKeyString)
|
||||
if !ok {
|
||||
return Identity{}
|
||||
}
|
||||
|
||||
id, _ = v.(Identity)
|
||||
|
||||
return id
|
||||
}
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
package middleware
|
||||
|
||||
import (
|
||||
"ruben/inventory2/internal/server/response"
|
||||
)
|
||||
|
||||
func Chain(ms ...response.Middleware) response.Middleware {
|
||||
return func(fn response.HandlerFunc) response.HandlerFunc {
|
||||
res := fn
|
||||
for _, m := range ms {
|
||||
prev := res
|
||||
res = m(prev)
|
||||
}
|
||||
return res
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
package middleware
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path"
|
||||
"ruben/inventory2/internal/logging"
|
||||
"ruben/inventory2/internal/server/sse"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
type UpdateNotificationPublisher struct {
|
||||
log *logging.Logger
|
||||
sse *sse.Queue
|
||||
trimBasePath string
|
||||
basePathPattern string
|
||||
}
|
||||
|
||||
func NewUpdateNotificationPublisher(
|
||||
log *logging.Logger,
|
||||
sse *sse.Queue,
|
||||
) *UpdateNotificationPublisher {
|
||||
return &UpdateNotificationPublisher{
|
||||
log: log,
|
||||
sse: sse,
|
||||
}
|
||||
}
|
||||
|
||||
func (p *UpdateNotificationPublisher) Trim(pathPattern string) *UpdateNotificationPublisher {
|
||||
p2 := *p
|
||||
p2.trimBasePath = path.Join(p2.trimBasePath, pathPattern)
|
||||
return &p2
|
||||
}
|
||||
|
||||
func (p *UpdateNotificationPublisher) Group(pathPattern string) *UpdateNotificationPublisher {
|
||||
p2 := *p
|
||||
p2.basePathPattern = path.Join(p2.basePathPattern, pathPattern)
|
||||
return &p2
|
||||
}
|
||||
|
||||
func (p *UpdateNotificationPublisher) Publish(pathPattern string) gin.HandlerFunc {
|
||||
trimBasePathSegs := getPathSegments(p.trimBasePath)
|
||||
trimmedBasePathPattern := path.Join(p.basePathPattern, pathPattern)
|
||||
trimmedBasePathPatternSegs := getPathSegments(trimmedBasePathPattern)
|
||||
fullBasePathPatternSegs := append(trimBasePathSegs, trimmedBasePathPatternSegs...)
|
||||
|
||||
return func(c *gin.Context) {
|
||||
reqPathSegs := getPathSegments(c.Request.URL.Path)
|
||||
|
||||
c.Next()
|
||||
if len(c.Errors) > 0 {
|
||||
return
|
||||
}
|
||||
if len(reqPathSegs) < len(fullBasePathPatternSegs) {
|
||||
p.log.Errorf("req path is less that the full path: %v, %v", c.Request.URL.Path, path.Join(p.trimBasePath, p.basePathPattern, pathPattern))
|
||||
return
|
||||
}
|
||||
|
||||
// if the path is a subpath, then emit events along the subpath.
|
||||
|
||||
for i, s := range trimmedBasePathPatternSegs {
|
||||
if isWildcard := s[0] == ':'; isWildcard {
|
||||
continue
|
||||
}
|
||||
|
||||
rs := reqPathSegs[i]
|
||||
if isSubpath := s != rs; isSubpath {
|
||||
continue
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
topEventParts := reqPathSegs[len(trimBasePathSegs):len(fullBasePathPatternSegs)]
|
||||
topEvent := strings.Join(topEventParts, "_")
|
||||
|
||||
events := make([]string, len(reqPathSegs)-len(fullBasePathPatternSegs)+1)
|
||||
events[0] = topEvent
|
||||
parentEvent := topEvent
|
||||
for i, s := range reqPathSegs[len(fullBasePathPatternSegs):] {
|
||||
e := parentEvent + "_" + s
|
||||
events[i+1] = e
|
||||
parentEvent = e
|
||||
}
|
||||
|
||||
for _, e := range events {
|
||||
go func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
|
||||
defer cancel()
|
||||
|
||||
if err := p.sse.Send(ctx, e, nil); err != nil {
|
||||
p.log.Errorf("failed to send sse event to listener: %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func getPathSegments(p string) []string {
|
||||
p = path.Clean(p)
|
||||
if p == "" || p == "." || p == "/" {
|
||||
return nil
|
||||
}
|
||||
|
||||
return strings.Split(strings.Trim(p, "/"), "/")
|
||||
}
|
||||
@@ -15,23 +15,29 @@ import (
|
||||
"ruben/inventory2/internal/logging"
|
||||
accounts_api "ruben/inventory2/internal/server/api/accounts"
|
||||
auth_api "ruben/inventory2/internal/server/api/auth"
|
||||
sse_api "ruben/inventory2/internal/server/api/sse"
|
||||
templates_api "ruben/inventory2/internal/server/api/templates"
|
||||
"ruben/inventory2/internal/server/api/webhooks"
|
||||
etsy_webhooks "ruben/inventory2/internal/server/api/webhooks/etsy"
|
||||
"ruben/inventory2/internal/server/middleware"
|
||||
"ruben/inventory2/internal/server/sse"
|
||||
)
|
||||
|
||||
// r gin.IRouter, // TODO: use this everywhere
|
||||
|
||||
func Router(
|
||||
ctx context.Context,
|
||||
type Router struct {
|
||||
*gin.Engine
|
||||
sse *sse.Queue
|
||||
}
|
||||
|
||||
func NewRouter(
|
||||
logger *logging.Logger,
|
||||
contentDir string,
|
||||
rawEvents *raw_events.Store,
|
||||
accts *accounts.Store,
|
||||
etsy *etsy_platform.Platform,
|
||||
auth *authentication.Authenticator,
|
||||
) *gin.Engine {
|
||||
) *Router {
|
||||
r := gin.Default()
|
||||
|
||||
// TODO: shouldn't this ACTUALLY be a middleware?
|
||||
@@ -74,21 +80,37 @@ func Router(
|
||||
authMiddleware,
|
||||
)
|
||||
|
||||
// sse setup
|
||||
|
||||
sq := sse.NewQueue()
|
||||
unp := middleware.NewUpdateNotificationPublisher(
|
||||
logger.WithGroup("update.notification.publisher"),
|
||||
sq,
|
||||
)
|
||||
|
||||
// api endpoints
|
||||
|
||||
api := r.Group("/api")
|
||||
apiLogger := logger.WithGroup("/api")
|
||||
unp = unp.Trim("/api")
|
||||
|
||||
auth_api.Routes(
|
||||
api.Group("/auth"),
|
||||
logger.WithGroup("/auth"),
|
||||
apiLogger.WithGroup("/auth"),
|
||||
auth,
|
||||
)
|
||||
accounts_api.Routes(
|
||||
api.Group("/accounts"),
|
||||
logger.WithGroup("/accounts"),
|
||||
accts,
|
||||
sse_api.Routes(
|
||||
api.Group("/events"),
|
||||
apiLogger.WithGroup("/events"),
|
||||
sq,
|
||||
authMiddleware,
|
||||
)
|
||||
accounts_api.Routes(
|
||||
api.Group("/accounts", authMiddleware.AuthenticateAndAddIdentityGin()),
|
||||
apiLogger.WithGroup("/accounts"),
|
||||
accts,
|
||||
unp.Group("/accounts"),
|
||||
)
|
||||
|
||||
// api webhooks (TODO: make a router for these)
|
||||
|
||||
@@ -104,7 +126,14 @@ func Router(
|
||||
},
|
||||
)
|
||||
|
||||
return r
|
||||
return &Router{
|
||||
Engine: r,
|
||||
sse: sq,
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Router) RunSSE(ctx context.Context) error {
|
||||
return r.sse.Start(ctx)
|
||||
}
|
||||
|
||||
func fileServer(urlPrefix, dir string, beforeServe func(c *gin.Context)) gin.HandlerFunc {
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
package sse
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type (
|
||||
Queue struct {
|
||||
in chan Event
|
||||
out map[int]chan Event
|
||||
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
|
||||
lock sync.Mutex
|
||||
prevID int
|
||||
}
|
||||
|
||||
Event struct {
|
||||
Type string
|
||||
Data []byte
|
||||
}
|
||||
)
|
||||
|
||||
func NewQueue() *Queue {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
return &Queue{
|
||||
in: make(chan Event),
|
||||
out: make(map[int]chan Event),
|
||||
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
}
|
||||
}
|
||||
|
||||
func (q *Queue) Start(ctx context.Context) error {
|
||||
defer q.cancel()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
// we're done piping events
|
||||
return nil
|
||||
case e := <-q.in:
|
||||
// share event will all subscribers
|
||||
q.lock.Lock()
|
||||
for _, out := range q.out {
|
||||
out <- e
|
||||
}
|
||||
q.lock.Unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (q *Queue) Listen(ctx context.Context, fn func(context.Context, *Event) error) error {
|
||||
// create new out pipe and append it to the queue
|
||||
|
||||
out := make(chan Event, 1)
|
||||
|
||||
q.lock.Lock()
|
||||
q.prevID += 1
|
||||
id := q.prevID
|
||||
q.out[id] = out
|
||||
q.lock.Unlock()
|
||||
|
||||
// delete the pipe when done listening
|
||||
defer func() {
|
||||
q.lock.Lock()
|
||||
delete(q.out, id)
|
||||
q.lock.Unlock()
|
||||
}()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-q.ctx.Done():
|
||||
// queue is shut down
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
// done listening to events
|
||||
return nil
|
||||
case e := <-out:
|
||||
// pass event to the caller
|
||||
if err := fn(ctx, &e); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (q *Queue) Send(ctx context.Context, eventType string, data []byte) error {
|
||||
select {
|
||||
case <-q.ctx.Done():
|
||||
// the queue has closed
|
||||
return fmt.Errorf("event queue closed: %w", q.ctx.Err())
|
||||
case <-ctx.Done():
|
||||
// sender ran out of time
|
||||
return fmt.Errorf("provided context canceled: %w", ctx.Err())
|
||||
// push the event onto the queue
|
||||
case q.in <- Event{
|
||||
Type: eventType,
|
||||
Data: data,
|
||||
}:
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Event) Write(w http.ResponseWriter) {
|
||||
fmt.Fprintf(
|
||||
w,
|
||||
"event: %s\ndata: %s\n\n",
|
||||
e.Type,
|
||||
bytes.ReplaceAll(e.Data, []byte("\n"), []byte(" ")),
|
||||
)
|
||||
w.(http.Flusher).Flush()
|
||||
}
|
||||
Reference in New Issue
Block a user