amazon events published as server side event

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"
 
@@ -13,14 +15,13 @@ import (
 
 type (
 	Mocks struct {
-		log *logging.Logger
-		db  *pgxpool.Pool
+		log      *logging.Logger
+		db       *pgxpool.Pool
+		listener MockEventListener
 	}
 
-	event struct {
-		ShopID         string
-		EventTimestamp time.Time
-		EventID        string
+	MockEventListener interface {
+		Notify(context.Context, raw_events.Event) error
 	}
 )
 
@@ -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,
-						event_id AS eventid,
-						event_timestamp AS eventtimestamp
+						shop_id,
+						event_id,
+						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
 }