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

		// notifyRetryAfter is how long an event can sit "notified" (handed
		// to the listener) without an ack before the dispatcher notifies
		// it again. Deliberately a field, not a stored notify_again_at
		// column, so the cadence can vary (or be tuned in tests) without
		// touching any row.
		notifyRetryAfter time.Duration

		// pollInterval is the fallback cadence ProcessEvents' main loop
		// checks for unprocessed/retry-due events on its own, independent
		// of Postgres NOTIFY - a safety net for events that end up in
		// mock.shop_amazon_events without ever going through the
		// mock.raw_shop_events insert+trigger path that fires NOTIFY.
		pollInterval time.Duration

		ready     chan struct{}
		readyOnce sync.Once
	}

	// MockEventListener is handed events as they arrive. Notify should
	// return promptly - any real work (including calling out to another
	// system) can happen after it returns - but it must eventually call
	// ack exactly once, when the event has been fully handled. Until ack
	// is called, the dispatcher considers the event merely "notified"
	// and will call Notify again after notifyRetryAfter, so ack may be
	// called more than once total across retries; only the first call
	// has any effect; further calls are safe (see (*Mocks).ack).
	MockEventListener interface {
		Notify(ctx context.Context, e raw_events.Event, ack func(context.Context) error) error
	}
)

const (
	eventChannelName = "mock_shop_amazon_event_inserted"

	defaultNotifyRetryAfter = 30 * time.Second
	defaultPollInterval     = time.Minute

	// backoff for reconnecting the LISTEN connection after it fails for a
	// reason other than shutdown (e.g. a dropped connection, a Postgres
	// restart). Doubles on each consecutive failure, capped, and resets
	// once a reconnect actually succeeds.
	initialListenReconnectBackoff = 1 * time.Second
	maxListenReconnectBackoff     = 30 * time.Second
)

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

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

// WithNotifyRetryAfter overrides how long an event can sit "notified"
// without an ack before being notified again. Mainly useful for tests that
// don't want to wait out the default.
func (m *Mocks) WithNotifyRetryAfter(d time.Duration) *Mocks {
	m.notifyRetryAfter = d
	return m
}

// WithPollInterval overrides how often ProcessEvents checks for
// unprocessed/retry-due events on its own, independent of NOTIFY. Mainly
// useful for tests that don't want to wait out the default.
func (m *Mocks) WithPollInterval(d time.Duration) *Mocks {
	m.pollInterval = d
	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)
	backoff := initialListenReconnectBackoff

	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:
			var stop bool
			if notifCh, errCh, backoff, stop = m.reconnectOrStop(ctx, err, backoff); stop {
				return nil
			}

		case _, ok := <-notifCh:
			if !ok {
				var stop bool
				if notifCh, errCh, backoff, stop = m.reconnectOrStop(ctx, <-errCh, backoff); stop {
					return nil
				}
				continue
			}
			m.log.Debug("woke up: notification received")

		case <-time.After(m.pollInterval):
			m.log.Debug("woke up: poll interval elapsed")
		}
	}
}

// reconnectOrStop handles a LISTEN-connection failure. A nil listenErr (or
// ctx already being done) means this is an ordinary shutdown, not a
// failure - stop is true and the caller should return. Otherwise it logs a
// warning, waits out backoff, and re-establishes LISTEN: on success the
// backoff resets to its initial value for next time; on immediate failure
// (e.g. the pool itself is unreachable) it doubles, capped, so repeated
// failures back off rather than hot-looping.
func (m *Mocks) reconnectOrStop(
	ctx context.Context,
	listenErr error,
	backoff time.Duration,
) (notifCh <-chan struct{}, errCh <-chan error, nextBackoff time.Duration, stop bool) {
	if listenErr == nil || ctx.Err() != nil {
		return nil, nil, backoff, true
	}

	m.log.Warn("lost connection while listening for notifications; reconnecting", "error", listenErr, "retry_in", backoff)

	select {
	case <-ctx.Done():
		return nil, nil, backoff, true
	case <-time.After(backoff):
	}

	notifCh, errCh = m.listenForNotifications(ctx)
	if notifCh != nil {
		return notifCh, errCh, initialListenReconnectBackoff, false
	}

	nextBackoff = backoff * 2
	if nextBackoff > maxListenReconnectBackoff {
		nextBackoff = maxListenReconnectBackoff
	}
	return notifCh, errCh, nextBackoff, false
}

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)
			}

			select {
			case ch <- struct{}{}:
			case <-ctx.Done():
				return nil
			}
		}
	}()

	return ch, errCh
}

// processUnprocessedEvents finds every event that's either brand new or
// has been "notified" for longer than notifyRetryAfter without an ack, and
// (re)dispatches each to the listener. It never blocks on the listener:
// notified_at is committed first, and only then - after the transaction
// that recorded it has actually committed - is the listener told, from a
// goroutine that isn't tied to this function's lifetime.
func (m *Mocks) processUnprocessedEvents(ctx context.Context) error {
	for done := false; !done; {
		var toDispatch []raw_events.Event

		err := pgx.BeginFunc(ctx, m.db, func(tx pgx.Tx) error {
			rows, err := tx.Query(
				ctx,
				`
					SELECT
						shop_id,
						event_id,
						event_timestamp
					FROM
						mock.shop_amazon_events
					WHERE
						processed_at IS NULL
						AND (notified_at IS NULL OR notified_at < @retry_after)
					ORDER BY
						shop_id, event_timestamp
					LIMIT
						100
				`,
				pgx.NamedArgs{
					"retry_after": time.Now().Add(-m.notifyRetryAfter),
				},
			)
			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 {
				ev := raw_events.Event{
					Platform:       string(accounts.Amazon),
					StoreID:        e.Shop_id,
					EventID:        e.Event_id,
					EventTimestamp: e.Event_timestamp,
				}

				if err := markNotified(ctx, tx, ev); err != nil {
					return fmt.Errorf("failed to mark event notified: %w", err)
				}

				toDispatch = append(toDispatch, ev)
			}

			if done = len(evts) == 0; !done {
				m.log.Infof("notified listener for %d events", len(evts))
			}

			return nil
		})
		if err != nil {
			return err
		}

		for _, e := range toDispatch {
			m.dispatch(ctx, e)
		}
	}

	return nil
}

// dispatch hands e to the configured listener without blocking. The
// listener may call ack synchronously (e.g. inline once its own work is
// done) or from elsewhere entirely, arbitrarily later - ack does its own
// independent, idempotent write, so it's never at risk of racing (or being
// silently lost to) this call's context being cancelled.
func (m *Mocks) dispatch(ctx context.Context, e raw_events.Event) {
	if m.listener == nil {
		return
	}

	go func() {
		if err := m.listener.Notify(ctx, e, func(ackCtx context.Context) error {
			return ack(ackCtx, m.db, e)
		}); err != nil {
			m.log.Errorf("error incurred by event listener: %v", err)
		}
	}()
}

func markNotified(ctx context.Context, tx pgx.Tx, e raw_events.Event) error {
	_, err := tx.Exec(
		ctx,
		`
			UPDATE
				mock.shop_amazon_events
			SET
				notified_at = NOW()
			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,
		},
	)
	return err
}

// ack records that e has been fully handled. It's idempotent - calling it
// more than once (e.g. because a retried notification also eventually acks)
// just leaves processed_at at whenever it was first set.
func ack(ctx context.Context, db *pgxpool.Pool, e raw_events.Event) error {
	_, err := db.Exec(
		ctx,
		`
			UPDATE
				mock.shop_amazon_events
			SET
				processed_at = NOW()
			WHERE
				shop_id = @shop_id
				AND event_id = @event_id
				AND event_timestamp = @event_timestamp
				AND processed_at IS NULL
		`,
		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)
	}
	return nil
}