184 lines
3.5 KiB
Go
184 lines
3.5 KiB
Go
package amazon
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"ruben/inventory2/logging"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
type (
|
|
Mocks struct {
|
|
log *logging.Logger
|
|
db *pgxpool.Pool
|
|
}
|
|
|
|
event struct {
|
|
ShopID string
|
|
EventTimestamp time.Time
|
|
EventID string
|
|
}
|
|
)
|
|
|
|
const (
|
|
eventChannelName = "mock_shop_amazon_event_inserted"
|
|
)
|
|
|
|
func NewMocks(log *logging.Logger, db *pgxpool.Pool) *Mocks {
|
|
return &Mocks{
|
|
log: log,
|
|
db: db,
|
|
}
|
|
}
|
|
|
|
func (m *Mocks) ProcessEvents(ctx context.Context) error {
|
|
notifCh, errCh := m.listenForNotifications(ctx)
|
|
|
|
for {
|
|
m.log.Debug("processing events")
|
|
if err := m.processUnprocessedEvents(ctx); err != nil {
|
|
if errors.Is(err, context.Canceled) {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("error occurred while processing events: %w", err)
|
|
}
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil
|
|
case err := <-errCh:
|
|
return err
|
|
case _, ok := <-notifCh:
|
|
if !ok {
|
|
return <-errCh
|
|
}
|
|
case <-time.After(time.Minute):
|
|
}
|
|
}
|
|
}
|
|
|
|
func (m *Mocks) listenForNotifications(ctx context.Context) (<-chan struct{}, <-chan error) {
|
|
errCh := make(chan error, 1)
|
|
|
|
pc, err := m.db.Acquire(ctx)
|
|
if err != nil {
|
|
errCh <- fmt.Errorf("failed to acquire a connection: %w", err)
|
|
return nil, errCh
|
|
}
|
|
|
|
conn := pc.Conn()
|
|
|
|
if _, err := conn.Exec(ctx, fmt.Sprintf("LISTEN %s", eventChannelName)); err != nil {
|
|
errCh <- fmt.Errorf("failed to start listening for notifications: %w", err)
|
|
return nil, errCh
|
|
}
|
|
|
|
ch := make(chan struct{})
|
|
|
|
go func() (err error) {
|
|
defer func() {
|
|
pc.Release()
|
|
if err != nil {
|
|
errCh <- err
|
|
}
|
|
close(ch)
|
|
close(errCh)
|
|
}()
|
|
|
|
for {
|
|
if _, err := conn.WaitForNotification(ctx); err != nil {
|
|
if errors.Is(err, context.Canceled) {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("error occurred while waiting for the next notification: %w", err)
|
|
}
|
|
|
|
ch <- struct{}{}
|
|
}
|
|
}()
|
|
|
|
return ch, errCh
|
|
}
|
|
|
|
func (m *Mocks) processUnprocessedEvents(ctx context.Context) error {
|
|
for done := false; !done; {
|
|
err := pgx.BeginFunc(ctx, m.db, func(tx pgx.Tx) error {
|
|
rows, err := m.db.Query(
|
|
ctx,
|
|
`
|
|
SELECT
|
|
shop_id AS shopid,
|
|
event_id AS eventid,
|
|
event_timestamp AS eventtimestamp
|
|
FROM
|
|
mock.shop_amazon_events
|
|
WHERE
|
|
NOT processed
|
|
ORDER BY
|
|
shop_id, event_timestamp
|
|
LIMIT
|
|
100
|
|
`,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to to perform query: %w", err)
|
|
}
|
|
|
|
evts, err := pgx.CollectRows(rows, pgx.RowToStructByNameLax[event])
|
|
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 {
|
|
return fmt.Errorf("failed to process event: %w", err)
|
|
}
|
|
}
|
|
|
|
if done = len(evts) == 0; !done {
|
|
m.log.Infof("processed %d events", len(evts))
|
|
}
|
|
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (m *Mocks) processEvent(ctx context.Context, tx pgx.Tx, e event) error {
|
|
m.log.Debugf("processing (mock) event: %#v", e)
|
|
|
|
_, err := tx.Exec(
|
|
ctx,
|
|
`
|
|
UPDATE
|
|
mock.shop_amazon_events
|
|
SET
|
|
processed = true,
|
|
processed_successfully = true
|
|
WHERE
|
|
shop_id = @shop_id
|
|
AND event_id = @event_id
|
|
AND event_timestamp = @event_timestamp
|
|
`,
|
|
pgx.NamedArgs{
|
|
"shop_id": e.ShopID,
|
|
"event_id": e.EventID,
|
|
"event_timestamp": e.EventTimestamp,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to perform query: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|