amazon events published as server side event

server/server.go

@@ -110,6 +110,10 @@ func (r *Router) RunSSE(ctx context.Context) error {
 	return r.sse.Start(ctx)
 }
 
+func (r *Router) GetSSEQueue() *sse.Queue {
+	return r.sse
+}
+
 func fileServer(urlPrefix, dir string, beforeServe func(c *gin.Context)) gin.HandlerFunc {
 	scfs := http.StripPrefix(urlPrefix, http.FileServer(http.Dir(dir)))
 	return func(c *gin.Context) {

server/sse/db_event_publisher.go

@@ -0,0 +1,44 @@
+package sse
+
+import (
+	"context"
+	"fmt"
+	"ruben/inventory2/domains/raw_events"
+)
+
+type (
+	DBEventPublisher struct {
+		queue        sender
+		getAccountID func(context.Context, raw_events.Event) (int64, error)
+		getEventType func(acctID int64, e raw_events.Event) (string, error)
+	}
+)
+
+func (q *Queue) NewDBEventPublisher(
+	getAccountID func(context.Context, raw_events.Event) (int64, error),
+	getEventType func(acctID int64, e raw_events.Event) (string, error),
+) *DBEventPublisher {
+	return &DBEventPublisher{
+		queue:        q,
+		getAccountID: getAccountID,
+		getEventType: getEventType,
+	}
+}
+
+func (p *DBEventPublisher) Notify(ctx context.Context, e raw_events.Event) error {
+	acctID, err := p.getAccountID(ctx, e)
+	if err != nil {
+		return fmt.Errorf("failed to get account id: %w", err)
+	}
+
+	et, err := p.getEventType(acctID, e)
+	if err != nil {
+		return fmt.Errorf("failed to compute event type: %w", err)
+	}
+
+	if err := p.queue.Send(ctx, StandardEvent(acctID, et)); err != nil {
+		return fmt.Errorf("failed to send sse event to listener: %w", err)
+	}
+
+	return nil
+}

server/sse/publisher.go

@@ -22,11 +22,6 @@ type (
 		trimBasePath    string
 		basePathPattern string
 	}
-
-	// sender is satisfied by *Queue
-	sender interface {
-		Send(ctx context.Context, e Event) error
-	}
 )
 
 func (q *Queue) NewUpdateNotificationPublisher(
@@ -110,11 +105,7 @@ func (p *UpdateNotificationPublisher) Publish(pathPattern string) gin.HandlerFun
 				ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
 				defer cancel()
 
-				if err := p.queue.Send(ctx, Event{
-					AccountID: acctID,
-					Type:      e,
-					Data:      []byte(fmt.Sprintf(`{"eventType": %q}`, e)),
-				}); err != nil {
+				if err := p.queue.Send(ctx, StandardEvent(acctID, e)); err != nil {
 					p.log.Errorf("failed to send sse event to listener: %v", err)
 				}
 			}()

server/sse/queue.go

@@ -124,6 +124,14 @@ func (q *Queue) Send(ctx context.Context, e Event) error {
 	return nil
 }
 
+func StandardEvent(acctID int64, eventType string) Event {
+	return Event{
+		AccountID: acctID,
+		Type:      eventType,
+		Data:      []byte(fmt.Sprintf(`{"eventType": %q}`, eventType)),
+	}
+}
+
 func (e *Event) Write(w http.ResponseWriter) {
 	fmt.Fprintf(
 		w,

server/sse/sender.go

@@ -0,0 +1,10 @@
+package sse
+
+import "context"
+
+type (
+	// sender is satisfied by *Queue
+	sender interface {
+		Send(ctx context.Context, e Event) error
+	}
+)