angel/inventory-plus-plus
1
0
Fork 0
Code Issues 34 Pull Requests Actions Packages Projects 4 Releases Wiki Activity

amazon events published as server side event

Browse Source
This commit is contained in:
angel
2026-08-20 00:35:13 -06:00
parent 16ecf2dc66
commit 4748eda41e
9 changed files with 186 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
domains/accounts/accounts.go
+38
View File
@@ -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
+34 -11
View File
@@ -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 {
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
}
main.go
+32 -11
View File
@@ -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
}
server/server.go
+4
View File
@@ -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
+44
View File
@@ -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
+1 -10
View File
@@ -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
+8
View File
@@ -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
+10
View File
@@ -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
+13 -1
View File
@@ -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(/-.*/, ''),