feat(dashboard-api): supabase auth users sync background runner

packages/dashboard-api/go.mod

@@ -23,6 +23,7 @@ require (
 	github.com/jackc/pgx/v5 v5.7.5
 	github.com/oapi-codegen/gin-middleware v1.0.2
 	github.com/oapi-codegen/runtime v1.1.1
+	github.com/stretchr/testify v1.11.1
 	go.uber.org/zap v1.27.1
 )
 
@@ -117,7 +118,6 @@ require (
 	github.com/shirou/gopsutil/v4 v4.25.9 // indirect
 	github.com/shopspring/decimal v1.4.0 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
-	github.com/stretchr/testify v1.11.1 // indirect
 	github.com/testcontainers/testcontainers-go v0.40.0 // indirect
 	github.com/testcontainers/testcontainers-go/modules/postgres v0.39.0 // indirect
 	github.com/tklauser/go-sysconf v0.3.15 // indirect

packages/dashboard-api/internal/cfg/model.go

@@ -1,6 +1,8 @@
 package cfg
 
 import (
+	"time"
+
 	"github.com/caarlos0/env/v11"
 )
 
@@ -12,6 +14,12 @@ type Config struct {
 
 	AuthDBConnectionString            string `env:"AUTH_DB_CONNECTION_STRING"`
 	AuthDBReadReplicaConnectionString string `env:"AUTH_DB_READ_REPLICA_CONNECTION_STRING"`
+
+	SupabaseAuthUserSyncEnabled      bool          `env:"SUPABASE_AUTH_USER_SYNC_ENABLED"       envDefault:"false"`
+	SupabaseAuthUserSyncBatchSize    int32         `env:"SUPABASE_AUTH_USER_SYNC_BATCH_SIZE"    envDefault:"50"`
+	SupabaseAuthUserSyncPollInterval time.Duration `env:"SUPABASE_AUTH_USER_SYNC_POLL_INTERVAL" envDefault:"2s"`
+	SupabaseAuthUserSyncLockTimeout  time.Duration `env:"SUPABASE_AUTH_USER_SYNC_LOCK_TIMEOUT"  envDefault:"2m"`
+	SupabaseAuthUserSyncMaxAttempts  int32         `env:"SUPABASE_AUTH_USER_SYNC_MAX_ATTEMPTS"  envDefault:"20"`
 }
 
 func Parse() (Config, error) {

packages/dashboard-api/internal/supabaseauthusersync/config.go

@@ -0,0 +1,21 @@
+package supabaseauthusersync
+
+import "time"
+
+type Config struct {
+	Enabled      bool
+	BatchSize    int32
+	PollInterval time.Duration
+	LockTimeout  time.Duration
+	MaxAttempts  int32
+}
+
+func DefaultConfig() Config {
+	return Config{
+		Enabled:      false,
+		BatchSize:    50,
+		PollInterval: 2 * time.Second,
+		LockTimeout:  2 * time.Minute,
+		MaxAttempts:  20,
+	}
+}

packages/dashboard-api/internal/supabaseauthusersync/processor.go

@@ -0,0 +1,135 @@
+package supabaseauthusersync
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"runtime/debug"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"go.uber.org/zap"
+
+	"github.com/e2b-dev/infra/packages/shared/pkg/logger"
+)
+
+type processorStore interface {
+	Ack(ctx context.Context, id int64) error
+	Retry(ctx context.Context, id int64, backoff time.Duration, lastError string) error
+	DeadLetter(ctx context.Context, id int64, lastError string) error
+	GetAuthUser(ctx context.Context, userID uuid.UUID) (*AuthUser, error)
+	UpsertPublicUser(ctx context.Context, id uuid.UUID, email string) error
+	DeletePublicUser(ctx context.Context, id uuid.UUID) error
+}
+
+type Processor struct {
+	store       processorStore
+	maxAttempts int32
+	l           logger.Logger
+}
+
+func NewProcessor(store processorStore, maxAttempts int32, l logger.Logger) *Processor {
+	return &Processor{
+		store:       store,
+		maxAttempts: maxAttempts,
+		l:           l,
+	}
+}
+
+func (p *Processor) Process(ctx context.Context, item QueueItem) {
+	err := p.processOnce(ctx, item)
+
+	if err == nil {
+		if ackErr := p.store.Ack(ctx, item.ID); ackErr != nil {
+			p.l.Error(ctx, "failed to ack queue item",
+				zap.Int64("queue_item_id", item.ID),
+				zap.String("user_id", item.UserID.String()),
+				zap.Error(ackErr),
+			)
+		}
+
+		return
+	}
+
+	p.l.Warn(ctx, "failed to process queue item",
+		zap.Int64("queue_item_id", item.ID),
+		zap.String("user_id", item.UserID.String()),
+		zap.Int32("attempt", item.AttemptCount),
+		zap.Error(err),
+	)
+
+	if item.AttemptCount >= p.maxAttempts {
+		if dlErr := p.store.DeadLetter(ctx, item.ID, err.Error()); dlErr != nil {
+			p.l.Error(ctx, "failed to dead-letter queue item",
+				zap.Int64("queue_item_id", item.ID),
+				zap.Error(dlErr),
+			)
+		}
+
+		return
+	}
+
+	backoff := retryBackoff(item.AttemptCount)
+
+	if retryErr := p.store.Retry(ctx, item.ID, backoff, err.Error()); retryErr != nil {
+		p.l.Error(ctx, "failed to retry queue item",
+			zap.Int64("queue_item_id", item.ID),
+			zap.Error(retryErr),
+		)
+	}
+}
+
+func (p *Processor) processOnce(ctx context.Context, item QueueItem) (err error) {
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			p.l.Error(ctx, "panic while processing queue item",
+				zap.Int64("queue_item_id", item.ID),
+				zap.String("user_id", item.UserID.String()),
+				zap.String("panic", fmt.Sprint(recovered)),
+				zap.String("stack", string(debug.Stack())),
+			)
+
+			err = fmt.Errorf("panic while processing queue item: %v", recovered)
+		}
+	}()
+
+	return p.reconcile(ctx, item)
+}
+
+func (p *Processor) reconcile(ctx context.Context, item QueueItem) error {
+	authUser, err := p.store.GetAuthUser(ctx, item.UserID)
+
+	if errors.Is(err, pgx.ErrNoRows) {
+		if delErr := p.store.DeletePublicUser(ctx, item.UserID); delErr != nil {
+			return fmt.Errorf("delete public.users %s: %w", item.UserID, delErr)
+		}
+
+		return nil
+	}
+
+	if err != nil {
+		return fmt.Errorf("get auth.users %s: %w", item.UserID, err)
+	}
+
+	if err = p.store.UpsertPublicUser(ctx, authUser.ID, authUser.Email); err != nil {
+		return fmt.Errorf("upsert public.users %s: %w", authUser.ID, err)
+	}
+
+	return nil
+}
+
+func retryBackoff(attempt int32) time.Duration {
+	switch {
+	case attempt <= 1:
+		return 5 * time.Second
+	case attempt <= 3:
+		return 30 * time.Second
+	case attempt <= 6:
+		return 2 * time.Minute
+	case attempt <= 10:
+		return 5 * time.Minute
+	default:
+		return 15 * time.Minute
+	}
+}

packages/dashboard-api/internal/supabaseauthusersync/processor_test.go

@@ -0,0 +1,109 @@
+package supabaseauthusersync
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+
+	"github.com/e2b-dev/infra/packages/shared/pkg/logger"
+)
+
+type retryCall struct {
+	id        int64
+	backoff   time.Duration
+	lastError string
+}
+
+type deadLetterCall struct {
+	id        int64
+	lastError string
+}
+
+type fakeProcessorStore struct {
+	getAuthUserFn func(context.Context, uuid.UUID) (*AuthUser, error)
+
+	ackCalls        []int64
+	retryCalls      []retryCall
+	deadLetterCalls []deadLetterCall
+}
+
+func (s *fakeProcessorStore) Ack(_ context.Context, id int64) error {
+	s.ackCalls = append(s.ackCalls, id)
+	return nil
+}
+
+func (s *fakeProcessorStore) Retry(_ context.Context, id int64, backoff time.Duration, lastError string) error {
+	s.retryCalls = append(s.retryCalls, retryCall{
+		id:        id,
+		backoff:   backoff,
+		lastError: lastError,
+	})
+	return nil
+}
+
+func (s *fakeProcessorStore) DeadLetter(_ context.Context, id int64, lastError string) error {
+	s.deadLetterCalls = append(s.deadLetterCalls, deadLetterCall{
+		id:        id,
+		lastError: lastError,
+	})
+	return nil
+}
+
+func (s *fakeProcessorStore) GetAuthUser(ctx context.Context, userID uuid.UUID) (*AuthUser, error) {
+	return s.getAuthUserFn(ctx, userID)
+}
+
+func (s *fakeProcessorStore) UpsertPublicUser(_ context.Context, _ uuid.UUID, _ string) error {
+	return nil
+}
+
+func (s *fakeProcessorStore) DeletePublicUser(_ context.Context, _ uuid.UUID) error {
+	return nil
+}
+
+func TestProcessorProcessRetriesRecoveredPanic(t *testing.T) {
+	store := &fakeProcessorStore{
+		getAuthUserFn: func(context.Context, uuid.UUID) (*AuthUser, error) {
+			panic("boom")
+		},
+	}
+	processor := NewProcessor(store, 3, logger.NewNopLogger())
+	item := QueueItem{
+		ID:           1,
+		UserID:       uuid.New(),
+		AttemptCount: 1,
+	}
+
+	require.NotPanics(t, func() {
+		processor.Process(context.Background(), item)
+	})
+	require.Empty(t, store.ackCalls)
+	require.Len(t, store.retryCalls, 1)
+	require.Contains(t, store.retryCalls[0].lastError, "panic while processing queue item")
+	require.Empty(t, store.deadLetterCalls)
+}
+
+func TestProcessorProcessDeadLettersRecoveredPanicAtMaxAttempts(t *testing.T) {
+	store := &fakeProcessorStore{
+		getAuthUserFn: func(context.Context, uuid.UUID) (*AuthUser, error) {
+			panic("boom")
+		},
+	}
+	processor := NewProcessor(store, 3, logger.NewNopLogger())
+	item := QueueItem{
+		ID:           1,
+		UserID:       uuid.New(),
+		AttemptCount: 3,
+	}
+
+	require.NotPanics(t, func() {
+		processor.Process(context.Background(), item)
+	})
+	require.Empty(t, store.ackCalls)
+	require.Empty(t, store.retryCalls)
+	require.Len(t, store.deadLetterCalls, 1)
+	require.Contains(t, store.deadLetterCalls[0].lastError, "panic while processing queue item")
+}

packages/dashboard-api/internal/supabaseauthusersync/runner.go

@@ -0,0 +1,69 @@
+package supabaseauthusersync
+
+import (
+	"context"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/e2b-dev/infra/packages/shared/pkg/logger"
+)
+
+type Runner struct {
+	cfg       Config
+	store     *Store
+	processor *Processor
+	lockOwner string
+	l         logger.Logger
+}
+
+func NewRunner(cfg Config, store *Store, lockOwner string, l logger.Logger) *Runner {
+	return &Runner{
+		cfg:       cfg,
+		store:     store,
+		processor: NewProcessor(store, cfg.MaxAttempts, l),
+		lockOwner: lockOwner,
+		l:         l,
+	}
+}
+
+func (r *Runner) Run(ctx context.Context) error {
+	r.l.Info(ctx, "starting supabase auth user sync worker",
+		zap.String("lock_owner", r.lockOwner),
+		zap.Duration("poll_interval", r.cfg.PollInterval),
+		zap.Int32("batch_size", r.cfg.BatchSize),
+	)
+
+	ticker := time.NewTicker(r.cfg.PollInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			r.l.Info(ctx, "stopping supabase auth user sync worker")
+
+			return ctx.Err()
+		case <-ticker.C:
+			r.poll(ctx)
+		}
+	}
+}
+
+func (r *Runner) poll(ctx context.Context) {
+	items, err := r.store.ClaimBatch(ctx, r.lockOwner, r.cfg.LockTimeout, r.cfg.BatchSize)
+	if err != nil {
+		r.l.Error(ctx, "failed to claim queue batch", zap.Error(err))
+
+		return
+	}
+
+	if len(items) == 0 {
+		return
+	}
+
+	r.l.Debug(ctx, "claimed queue batch", zap.Int("count", len(items)))
+
+	for _, item := range items {
+		r.processor.Process(ctx, item)
+	}
+}