inventory-plus-plus/domains/amazon/mock.go

package amazon

import (
	"context"
	"errors"
	"fmt"
	"ruben/inventory2/domains/accounts"
	"ruben/inventory2/domains/raw_events"
	"ruben/inventory2/logging"
	"sync"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

type (
	Mocks struct {
		log      *logging.Logger
		db       *pgxpool.Pool
		listener MockEventListener

		ready     chan struct{}
		readyOnce sync.Once
	}

	MockEventListener interface {
		Notify(context.Context, raw_events.Event) error
	}
)

const (
	eventChannelName = "mock_shop_amazon_event_inserted"
)

func NewMocks(log *logging.Logger, db *pgxpool.Pool) *Mocks {
	return &Mocks{
		log:   log,
		db:    db,
		ready: make(chan struct{}),
	}
}

func (m *Mocks) SetListener(l MockEventListener) *Mocks {
	m.listener = l
	return m
}

// Ready returns a channel that's closed once ProcessEvents has registered
// its Postgres LISTEN and is actively watching for notifications. Callers
// that need to know the reactive path is live - tests in particular -
// should wait on this instead of guessing with a sleep.
func (m *Mocks) Ready() <-chan struct{} {
	return m.ready
}

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
	}

	m.readyOnce.Do(func() { close(m.ready) })

	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,
						event_id,
						event_timestamp
					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[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, 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)
				}
			}

			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 raw_events.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.StoreID,
			"event_id":        e.EventID,
			"event_timestamp": e.EventTimestamp,
		},
	)
	if err != nil {
		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
}