encore-test/profiles/login.go

package profiles

import (
	"context"
	"errors"
	"sync"
	"time"

	"encore.app/auth"
	"encore.dev/beta/errs"
	"encore.dev/pubsub"
	"encore.dev/storage/sqldb"
	"encore.dev/types/uuid"
)

const loginTimeout = 10 * time.Second

type LoginParams struct {
	UserEmail string `json:"user_email"`
	Password  string `json:"password" encore:"sensitive"`
}

type LoginResponse struct {
	Token     uuid.UUID `json:"token"`
	ExpiresAt time.Time `json:"expires_at"`
}

var loginRequests = pubsub.TopicRef[pubsub.Publisher[*auth.LoginRequested]](auth.LoginRequests)

// Login is the public entry point for authenticating a user. It first
// resolves the given email to a user ID via the local profiles database.
// The actual credential check lives in the auth service; rather than calling
// it directly, this hands the request off over Pub/Sub (publishing to
// auth.LoginRequests) and waits for the matching auth.LoginCompleted reply
// on auth.LoginResults before responding to the caller.
//
// NOTE: the wait is implemented with an in-process registry keyed by request
// ID, so the instance that publishes the request must be the one that
// receives the reply. That holds for local development (a single instance)
// but not for a horizontally scaled deployment, which would need a shared
// mechanism instead (e.g. a results table polled with retry_until).
//
//encore:api public method=POST path=/auth/login
func Login(ctx context.Context, p *LoginParams) (*LoginResponse, error) {
	var userID uuid.UUID
	err := db.QueryRow(ctx, `
		SELECT user_id FROM user_profiles WHERE email = $1
	`, p.UserEmail).Scan(&userID)
	if errors.Is(err, sqldb.ErrNoRows) {
		return nil, &errs.Error{Code: errs.Unauthenticated, Message: "invalid credentials"}
	}
	if err != nil {
		return nil, errs.WrapCode(err, errs.Internal, "failed to look up user")
	}

	requestID, err := uuid.NewV4()
	if err != nil {
		return nil, errs.WrapCode(err, errs.Internal, "failed to generate request id")
	}

	wait := loginWaiters.register(requestID)
	defer loginWaiters.cancel(requestID)

	if _, err := loginRequests.Publish(ctx, &auth.LoginRequested{
		RequestID: requestID,
		UserID:    userID,
		Password:  p.Password,
	}); err != nil {
		return nil, errs.WrapCode(err, errs.Internal, "failed to publish login request")
	}

	select {
	case result := <-wait:
		if !result.OK {
			return nil, &errs.Error{Code: errs.Unauthenticated, Message: "invalid credentials"}
		}
		return &LoginResponse{Token: result.Token, ExpiresAt: result.ExpiresAt}, nil
	case <-time.After(loginTimeout):
		return nil, &errs.Error{Code: errs.DeadlineExceeded, Message: "login timed out"}
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

var _ = pubsub.NewSubscription(
	auth.LoginResults, "deliver-to-waiting-login",
	pubsub.SubscriptionConfig[*auth.LoginCompleted]{
		Handler: func(ctx context.Context, ev *auth.LoginCompleted) error {
			loginWaiters.deliver(ev)
			return nil
		},
	},
)

// loginWaiters lets the synchronous Login handler above wait for the
// asynchronous LoginCompleted reply that matches its RequestID.
var loginWaiters = newWaiterRegistry()

type waiterRegistry struct {
	mu      sync.Mutex
	waiting map[uuid.UUID]chan *auth.LoginCompleted
}

func newWaiterRegistry() *waiterRegistry {
	return &waiterRegistry{waiting: make(map[uuid.UUID]chan *auth.LoginCompleted)}
}

func (r *waiterRegistry) register(id uuid.UUID) <-chan *auth.LoginCompleted {
	ch := make(chan *auth.LoginCompleted, 1)
	r.mu.Lock()
	r.waiting[id] = ch
	r.mu.Unlock()
	return ch
}

func (r *waiterRegistry) cancel(id uuid.UUID) {
	r.mu.Lock()
	delete(r.waiting, id)
	r.mu.Unlock()
}

func (r *waiterRegistry) deliver(ev *auth.LoginCompleted) {
	r.mu.Lock()
	ch, ok := r.waiting[ev.RequestID]
	if ok {
		delete(r.waiting, ev.RequestID)
	}
	r.mu.Unlock()
	if ok {
		ch <- ev
	}
}