amazon events published as server side event

domains/accounts/accounts.go

@@ -227,6 +227,44 @@ func (db *Store) GetAccountByEmail(ctx context.Context, email string) (Account,
 	}, nil
 }
 
+func (db *Store) GetAccountIDByMockPlatformAndShopID(ctx context.Context, platform Platform, shopID string) (int64, error) {
+	info, ok := getMockShopSchemaInfo(platform)
+	if !ok {
+		return 0, fmt.Errorf("unrecognized platform: %w", consts.ErrNotFound)
+	}
+
+	rows, err := db.db.Query(
+		ctx,
+		fmt.Sprintf(
+			`
+				SELECT
+					account_id
+				FROM
+					%s
+				WHERE
+					shop_id = @shop_id
+			`,
+			info.shopTable,
+		),
+		pgx.NamedArgs{
+			"shop_id": shopID,
+		},
+	)
+	if err != nil {
+		return 0, fmt.Errorf("failed to perform query: %w", err)
+	}
+
+	acctID, err := pgx.CollectExactlyOneRow(rows, pgx.RowTo[int64])
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return 0, consts.ErrNotFound
+		}
+		return 0, fmt.Errorf("failed to scan rows: %w", err)
+	}
+
+	return acctID, nil
+}
+
 func (db *Store) GetUserAndAccountByAccessToken(ctx context.Context, accessToken string) (OAuthUser, *Account, error) {
 	rows, err := db.db.Query(
 		ctx,

domains/amazon/mock.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"ruben/inventory2/domains/accounts"
+	"ruben/inventory2/domains/raw_events"
 	"ruben/inventory2/logging"
 	"time"
 
@@ -15,12 +17,11 @@ type (
 	Mocks struct {
 		log      *logging.Logger
 		db       *pgxpool.Pool
+		listener MockEventListener
 	}
 
-	event struct {
+	MockEventListener interface {
-		ShopID         string
+		Notify(context.Context, raw_events.Event) error
-		EventTimestamp time.Time
-		EventID        string
 	}
 )
 
@@ -35,6 +36,11 @@ func NewMocks(log *logging.Logger, db *pgxpool.Pool) *Mocks {
 	}
 }
 
+func (m *Mocks) SetListener(l MockEventListener) *Mocks {
+	m.listener = l
+	return m
+}
+
 func (m *Mocks) ProcessEvents(ctx context.Context) error {
 	notifCh, errCh := m.listenForNotifications(ctx)
 
@@ -111,9 +117,9 @@ func (m *Mocks) processUnprocessedEvents(ctx context.Context) error {
 				ctx,
 				`
 					SELECT
-						shop_id AS shopid,
+						shop_id,
-						event_id AS eventid,
+						event_id,
-						event_timestamp AS eventtimestamp
+						event_timestamp
 					FROM
 						mock.shop_amazon_events
 					WHERE
@@ -128,13 +134,22 @@ func (m *Mocks) processUnprocessedEvents(ctx context.Context) error {
 				return fmt.Errorf("failed to to perform query: %w", err)
 			}
 
-			evts, err := pgx.CollectRows(rows, pgx.RowToStructByNameLax[event])
+			evts, err := pgx.CollectRows(rows, pgx.RowToStructByNameLax[struct {
+				Shop_id         string
+				Event_id        string
+				Event_timestamp time.Time
+			}])
 			if err != nil {
 				return fmt.Errorf("failed to scan rows: %w", err)
 			}
 
 			for _, e := range evts {
-				if err := m.processEvent(ctx, tx, e); err != nil {
+				if err := m.processEvent(ctx, tx, raw_events.Event{
+					Platform:       string(accounts.Amazon),
+					StoreID:        e.Shop_id,
+					EventID:        e.Event_id,
+					EventTimestamp: e.Event_timestamp,
+				}); err != nil {
 					return fmt.Errorf("failed to process event: %w", err)
 				}
 			}
@@ -153,7 +168,7 @@ func (m *Mocks) processUnprocessedEvents(ctx context.Context) error {
 	return nil
 }
 
-func (m *Mocks) processEvent(ctx context.Context, tx pgx.Tx, e event) error {
+func (m *Mocks) processEvent(ctx context.Context, tx pgx.Tx, e raw_events.Event) error {
 	m.log.Debugf("processing (mock) event: %#v", e)
 
 	_, err := tx.Exec(
@@ -170,7 +185,7 @@ func (m *Mocks) processEvent(ctx context.Context, tx pgx.Tx, e event) error {
 				AND event_timestamp = @event_timestamp
 		`,
 		pgx.NamedArgs{
-			"shop_id":         e.ShopID,
+			"shop_id":         e.StoreID,
 			"event_id":        e.EventID,
 			"event_timestamp": e.EventTimestamp,
 		},
@@ -179,5 +194,13 @@ func (m *Mocks) processEvent(ctx context.Context, tx pgx.Tx, e event) error {
 		return fmt.Errorf("failed to perform query: %w", err)
 	}
 
+	if m.listener != nil {
+		go func() {
+			if err := m.listener.Notify(ctx, e); err != nil {
+				m.log.Errorf("error incurred by event listener: %v", err)
+			}
+		}()
+	}
+
 	return nil
 }

main.go

@@ -29,6 +29,7 @@ import (
 	"ruben/inventory2/domains/reports"
 	"ruben/inventory2/logging"
 	"ruben/inventory2/server"
+	"ruben/inventory2/server/sse"
 )
 
 const (
@@ -74,24 +75,40 @@ func runApp(ctx context.Context, logger *logging.Logger) error {
 		return fmt.Errorf("failed to initialize database connection pool: %w", err)
 	}
 
-	// start background processes
-
 	auth, err := authentication.New(ctx, connPool, logger.WithGroup("authenticator"))
 	if err != nil {
 		return fmt.Errorf("failed to construct authenticator: %w", err)
 	}
 
+	accts := accounts.NewStore(logger.WithGroup("accounts"), connPool)
+
+	// start http server
+
+	sseQueue, srvErrCh := runServer(ctx, logger, connPool, auth, accts)
+
+	// start background processes
+
 	authErrCh := runAuthProcesses(ctx, auth)
 
 	eventErrCh := runEventProcessing(
 		ctx,
-		amazon.NewMocks(logger.WithGroup("amazon"), connPool),
+		amazon.NewMocks(logger.WithGroup("amazon"), connPool).
+			SetListener(sseQueue.NewDBEventPublisher(
+				func(ctx context.Context, e raw_events.Event) (acctID int64, err error) {
+					p, err := accounts.NewPlatform(e.Platform)
+					if err != nil {
+						return 0, err
+					}
+
+					return accts.GetAccountIDByMockPlatformAndShopID(ctx, p, e.StoreID)
+				},
+				func(acctID int64, e raw_events.Event) (eventType string, err error) {
+					// TODO: fine tune the event type later (don't want a referesh on EVERY event)
+					return fmt.Sprintf("accounts_%d_simulations", acctID), nil
+				},
+			)),
 	)
 
-	// start http server
-
-	srvErrCh := runServer(ctx, logger, connPool, auth)
-
 	// wait for interrupt signal or unrecoverable failure, then shutdown
 
 	osSignalCh := make(chan os.Signal, 1)
@@ -184,9 +201,13 @@ func runEventProcessing(ctx context.Context, amz *amazon.Mocks) <-chan error {
 	return errCh
 }
 
-func runServer(ctx context.Context, logger *logging.Logger, connPool *pgxpool.Pool, auth *authentication.Authenticator) <-chan error {
+func runServer(
-	accts := accounts.NewStore(logger.WithGroup("accounts"), connPool)
+	ctx context.Context,
-
+	logger *logging.Logger,
+	connPool *pgxpool.Pool,
+	auth *authentication.Authenticator,
+	accts *accounts.Store,
+) (*sse.Queue, <-chan error) {
 	r := server.NewRouter(
 		logger.WithGroup("server"),
 		"./",
@@ -268,5 +289,5 @@ func runServer(ctx context.Context, logger *logging.Logger, connPool *pgxpool.Po
 		}
 	}()
 
-	return errCh
+	return r.GetSSEQueue(), errCh
 }

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{
+				if err := p.queue.Send(ctx, StandardEvent(acctID, e)); err != nil {
-					AccountID: acctID,
-					Type:      e,
-					Data:      []byte(fmt.Sprintf(`{"eventType": %q}`, 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
+	}
+)

templates/pages/accounts/{acctID.int64}/reports/mock.html.tmpl

@@ -11,12 +11,24 @@
 {{- end }}
 
 
+{{/* sse handler */}}
+<div
+	hidden
+	hx-trigger="sse:accounts_{{$acctID}}_simulations"
+	hx-get="/ui/accounts/{{$acctID}}/reports/mock"
+	hx-vals="js:{
+		platform: (new URLSearchParams(window.location.search)).get('platform'),
+		'shop-id': (new URLSearchParams(window.location.search)).get('shop-id'),
+	}"
+	hx-target="body"
+>
+</div>
+
 <label for="shop-selector">
 	Select a Shop
 </label>
 <select
 	id="shop-selector"
-	hx-swap=""
 	hx-get="/ui/accounts/{{$acctID}}/reports/mock"
 	hx-vals="js:{
 		platform: console.log(event.target.value.replace(/-.*/, '')) || event.target.value.replace(/-.*/, ''),