From 55a3694d8989d4e443efd96e0855ce491c3e88d5 Mon Sep 17 00:00:00 2001 From: Angel Beltran Date: Tue, 3 Mar 2026 22:08:58 -0700 Subject: [PATCH] amazon events published as server side event --- domains/accounts/accounts.go | 38 ++++++++++++++ domains/amazon/mock.go | 49 ++++++++++++++----- main.go | 43 +++++++++++----- server/server.go | 4 ++ server/sse/db_event_publisher.go | 44 +++++++++++++++++ server/sse/publisher.go | 11 +---- server/sse/queue.go | 8 +++ server/sse/sender.go | 10 ++++ .../{acctID.int64}/reports/mock.html.tmpl | 14 +++++- 9 files changed, 186 insertions(+), 35 deletions(-) create mode 100644 server/sse/db_event_publisher.go create mode 100644 server/sse/sender.go diff --git a/domains/accounts/accounts.go b/domains/accounts/accounts.go index 09ee8d0..eb13f29 100644 --- a/domains/accounts/accounts.go +++ b/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, diff --git a/domains/amazon/mock.go b/domains/amazon/mock.go index 5653ac8..24260db 100644 --- a/domains/amazon/mock.go +++ b/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 } diff --git a/main.go b/main.go index b9f2edd..2213fe8 100644 --- a/main.go +++ b/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 { - accts := accounts.NewStore(logger.WithGroup("accounts"), connPool) - +func runServer( + 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 } diff --git a/server/server.go b/server/server.go index 7b91d54..89ba9e4 100644 --- a/server/server.go +++ b/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) { diff --git a/server/sse/db_event_publisher.go b/server/sse/db_event_publisher.go new file mode 100644 index 0000000..ff66042 --- /dev/null +++ b/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 +} diff --git a/server/sse/publisher.go b/server/sse/publisher.go index 6ab25e4..45b57a5 100644 --- a/server/sse/publisher.go +++ b/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) } }() diff --git a/server/sse/queue.go b/server/sse/queue.go index 6e0ac5e..64fbb05 100644 --- a/server/sse/queue.go +++ b/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, diff --git a/server/sse/sender.go b/server/sse/sender.go new file mode 100644 index 0000000..3aa1fd7 --- /dev/null +++ b/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 + } +) diff --git a/templates/pages/accounts/{acctID.int64}/reports/mock.html.tmpl b/templates/pages/accounts/{acctID.int64}/reports/mock.html.tmpl index 1395f04..39f0681 100644 --- a/templates/pages/accounts/{acctID.int64}/reports/mock.html.tmpl +++ b/templates/pages/accounts/{acctID.int64}/reports/mock.html.tmpl @@ -11,12 +11,24 @@ {{- end }} +{{/* sse handler */}} + +