moved middleware event publisher to sse package

This commit is contained in:
2026-01-24 14:06:19 -07:00
parent 720c3fb4e0
commit a6267ac189
5 changed files with 33 additions and 20 deletions
+2 -1
View File
@@ -9,6 +9,7 @@ import (
"ruben/inventory2/internal/logging" "ruben/inventory2/internal/logging"
"ruben/inventory2/internal/server/middleware" "ruben/inventory2/internal/server/middleware"
"ruben/inventory2/internal/server/response" "ruben/inventory2/internal/server/response"
"ruben/inventory2/internal/server/sse"
"strconv" "strconv"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
@@ -23,7 +24,7 @@ func Routes(
r *gin.RouterGroup, r *gin.RouterGroup,
logger *logging.Logger, logger *logging.Logger,
accts *accounts.Store, accts *accounts.Store,
pub *middleware.UpdateNotificationPublisher, pub *sse.UpdateNotificationPublisher,
) { ) {
as := &accountSubrouter{ as := &accountSubrouter{
log: logger, log: logger,
+1 -1
View File
@@ -22,7 +22,7 @@ func Routes(
auth *middleware.Auth, auth *middleware.Auth,
sq *sse.Queue, sq *sse.Queue,
accts *accounts.Store, accts *accounts.Store,
unp *middleware.UpdateNotificationPublisher, unp *sse.UpdateNotificationPublisher,
rawEvents *raw_events.Store, rawEvents *raw_events.Store,
etsy *etsy_platform.Platform, etsy *etsy_platform.Platform,
) { ) {
+4 -2
View File
@@ -79,9 +79,11 @@ func NewRouter(
// sse setup // sse setup
sq := sse.NewQueue() sq := sse.NewQueue()
unp := middleware.NewUpdateNotificationPublisher( unp := sq.NewUpdateNotificationPublisher(
logger.WithGroup("update.notification.publisher"), logger.WithGroup("update.notification.publisher"),
sq, func(c *gin.Context) int64 {
return middleware.GetIdentity(c).Account.AccountID
},
).Trim("/api") ).Trim("/api")
api.Routes( api.Routes(
@@ -1,31 +1,40 @@
package middleware package sse
import ( import (
"context" "context"
"fmt" "fmt"
"path" "path"
"ruben/inventory2/internal/logging"
"ruben/inventory2/internal/server/sse"
"strings" "strings"
"time" "time"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
"ruben/inventory2/internal/logging"
) )
type UpdateNotificationPublisher struct { type (
UpdateNotificationPublisher struct {
log *logging.Logger log *logging.Logger
sse *sse.Queue queue sender
getAccountID func(*gin.Context) int64
trimBasePath string trimBasePath string
basePathPattern string basePathPattern string
} }
func NewUpdateNotificationPublisher( // sender is satisfied by *Queue
sender interface {
Send(ctx context.Context, e Event) error
}
)
func (q *Queue) NewUpdateNotificationPublisher(
log *logging.Logger, log *logging.Logger,
sse *sse.Queue, getAccountID func(*gin.Context) int64,
) *UpdateNotificationPublisher { ) *UpdateNotificationPublisher {
return &UpdateNotificationPublisher{ return &UpdateNotificationPublisher{
log: log, log: log,
sse: sse, queue: q,
getAccountID: getAccountID,
} }
} }
@@ -46,7 +55,8 @@ func (p *UpdateNotificationPublisher) Group(pathPattern string) *UpdateNotificat
return &p2 return &p2
} }
// Publish must be applied AFTER a middleware puts in the Identity into the gin.Context. // Publish constructs a gin middleware.
// It must used with and after middleware puts in the Identity into the gin.Context.
func (p *UpdateNotificationPublisher) Publish(pathPattern string) gin.HandlerFunc { func (p *UpdateNotificationPublisher) Publish(pathPattern string) gin.HandlerFunc {
trimBasePathSegs := getPathSegments(p.trimBasePath) trimBasePathSegs := getPathSegments(p.trimBasePath)
trimmedBasePathPattern := path.Join(p.basePathPattern, pathPattern) trimmedBasePathPattern := path.Join(p.basePathPattern, pathPattern)
@@ -92,13 +102,13 @@ func (p *UpdateNotificationPublisher) Publish(pathPattern string) gin.HandlerFun
parentEvent = e parentEvent = e
} }
acctID := GetIdentity(c).Account.AccountID acctID := p.getAccountID(c)
for _, e := range events { for _, e := range events {
go func() { go func() {
ctx, cancel := context.WithTimeout(context.Background(), time.Minute) ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel() defer cancel()
if err := p.sse.Send(ctx, sse.Event{ if err := p.queue.Send(ctx, Event{
AccountID: acctID, AccountID: acctID,
Type: e, Type: e,
Data: []byte(fmt.Sprintf(`{"eventType": %q}`, e)), Data: []byte(fmt.Sprintf(`{"eventType": %q}`, e)),