rohit/probo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor sign-in page and IAM service lifecycle

Browse Source 
Redesign the sign-in page to show email/password form inline
with OIDC provider buttons (with vendor icons) instead of
separate pages. Extract OIDCProvider type to its own file.

Replace errgroup with sync.WaitGroup + WithCancelCause for
graceful shutdown in IAM services. Refactor garbage collectors
to use functional options and time.Ticker instead of
time.After to avoid repeated allocations.

Signed-off-by: Bryan Frimin <bryan@getprobo.com>
This commit is contained in:
Bryan Frimin
2026-03-21 19:11:09 +01:00
parent 23084a72a2
commit 83468db034
11 changed files with 384 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
pkg/iam/oidc/gc.go
View File

@@ -28,22 +28,40 @@ const (
DefaultGarbageCollectionInterval = 1 * time.Hour
)
type GarbageCollector struct {
pg *pg.Client
interval time.Duration
logger *log.Logger
type (
GarbageCollector struct {
pg *pg.Client
interval time.Duration
logger *log.Logger
}
GarbageCollectorOption func(*GarbageCollector)
)
func WithGarbageCollectionInterval(interval time.Duration) GarbageCollectorOption {
return func(gc *GarbageCollector) {
gc.interval = interval
}
}
func NewGarbageCollector(
pg *pg.Client,
interval time.Duration,
pgClient *pg.Client,
logger *log.Logger,
opts ...GarbageCollectorOption,
) *GarbageCollector {
return &GarbageCollector{
pg: pg,
interval: interval,
logger: logger.Named("oidc.garbage_collector").With(log.Duration("interval", interval)),
gc := &GarbageCollector{
pg: pgClient,
interval: DefaultGarbageCollectionInterval,
logger: logger.Named("oidc.garbage_collector"),
}
for _, opt := range opts {
opt(gc)
}
gc.logger = gc.logger.With(log.Duration("interval", gc.interval))
return gc
}
func (gc *GarbageCollector) Run(ctx context.Context) error {
@@ -53,12 +71,15 @@ func (gc *GarbageCollector) Run(ctx context.Context) error {
gc.logger.ErrorCtx(ctx, "cannot run initial cleanup", log.Error(err))
}
ticker := time.NewTicker(gc.interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
gc.logger.InfoCtx(ctx, "oidc garbage collector shutting down")
return ctx.Err()
case <-time.After(gc.interval):
case <-ticker.C:
if err := gc.cleanup(ctx); err != nil {
gc.logger.ErrorCtx(ctx, "cannot run periodic cleanup", log.Error(err))
}
@@ -72,7 +93,8 @@ func (gc *GarbageCollector) cleanup(ctx context.Context) error {
return gc.pg.WithTx(
ctx,
func(tx pg.Conn) error {
deleted, err := coredata.DeleteExpiredOIDCStates(ctx, tx, now)
var state coredata.OIDCState
deleted, err := state.DeleteExpired(ctx, tx, now)
if err != nil {
return fmt.Errorf("cannot delete expired oidc states: %w", err)
}

36
pkg/iam/oidc/service.go
View File

@@ -215,29 +215,25 @@ func NewService(
}
func (s *Service) Run(ctx context.Context) error {
gc := NewGarbageCollector(s.pg, DefaultGarbageCollectionInterval, s.logger)
wg := sync.WaitGroup{}
ctx, cancel := context.WithCancelCause(ctx)
defer cancel(context.Canceled)
gcCtx, stopGC := context.WithCancel(ctx)
defer stopGC()
errCh := make(chan error, 1)
go func() {
errCh <- gc.Run(gcCtx)
}()
select {
case <-ctx.Done():
stopGC()
<-errCh
return ctx.Err()
case err := <-errCh:
if err != nil {
s.logger.ErrorCtx(ctx, "oidc garbage collector failed", log.Error(err))
return err
gcCtx, stopGC := context.WithCancel(context.WithoutCancel(ctx))
gc := NewGarbageCollector(s.pg, s.logger)
wg.Go(func() {
if err := gc.Run(gcCtx); err != nil {
cancel(fmt.Errorf("oidc garbage collector crashed: %w", err))
}
})
return nil
}
<-ctx.Done()
stopGC()
wg.Wait()
return context.Cause(ctx)
}
func (s *Service) IsProviderEnabled(provider coredata.OIDCProvider) bool {

39
pkg/iam/saml/gc.go
View File

@@ -34,20 +34,36 @@ type (
interval time.Duration
logger *log.Logger
}
GarbageCollectorOption func(*GarbageCollector)
)
func NewGarbageCollector(
pg *pg.Client,
interval time.Duration,
logger *log.Logger,
) *GarbageCollector {
return &GarbageCollector{
pg: pg,
interval: interval,
logger: logger.Named("saml.garbage_collector").With(log.Duration("interval", interval)),
func WithGarbageCollectionInterval(interval time.Duration) GarbageCollectorOption {
return func(gc *GarbageCollector) {
gc.interval = interval
}
}
func NewGarbageCollector(
pgClient *pg.Client,
logger *log.Logger,
opts ...GarbageCollectorOption,
) *GarbageCollector {
gc := &GarbageCollector{
pg: pgClient,
interval: DefaultGarbageCollectionInterval,
logger: logger.Named("saml.garbage_collector"),
}
for _, opt := range opts {
opt(gc)
}
gc.logger = gc.logger.With(log.Duration("interval", gc.interval))
return gc
}
func (gc *GarbageCollector) Run(ctx context.Context) error {
gc.logger.InfoCtx(ctx, "saml garbage collector starting")
@@ -55,12 +71,15 @@ func (gc *GarbageCollector) Run(ctx context.Context) error {
gc.logger.ErrorCtx(ctx, "cannot run initial cleanup", log.Error(err))
}
ticker := time.NewTicker(gc.interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
gc.logger.InfoCtx(ctx, "saml garbage collector shutting down")
return ctx.Err()
case <-time.After(gc.interval):
case <-ticker.C:
if err := gc.cleanup(ctx); err != nil {
gc.logger.ErrorCtx(ctx, "cannot run periodic cleanup", log.Error(err))
}

37
pkg/iam/saml/service.go
View File

@@ -24,6 +24,7 @@ import (
"fmt"
"net/url"
"strings"
"sync"
"time"
"github.com/crewjam/saml"
@@ -70,29 +71,25 @@ func NewService(
}
func (s *Service) Run(ctx context.Context) error {
gc := NewGarbageCollector(s.pg, DefaultGarbageCollectionInterval, s.logger)
wg := sync.WaitGroup{}
ctx, cancel := context.WithCancelCause(ctx)
defer cancel(context.Canceled)
gcCtx, stopGC := context.WithCancel(ctx)
defer stopGC()
errCh := make(chan error, 1)
go func() {
errCh <- gc.Run(gcCtx)
}()
select {
case <-ctx.Done():
stopGC()
<-errCh
return ctx.Err()
case err := <-errCh:
if err != nil {
s.logger.ErrorCtx(ctx, "saml garbage collector failed", log.Error(err))
return err
gcCtx, stopGC := context.WithCancel(context.WithoutCancel(ctx))
gc := NewGarbageCollector(s.pg, s.logger)
wg.Go(func() {
if err := gc.Run(gcCtx); err != nil {
cancel(fmt.Errorf("saml garbage collector crashed: %w", err))
}
})
return nil
}
<-ctx.Done()
stopGC()
wg.Wait()
return context.Cause(ctx)
}
func (s *Service) GenerateSpMetadata() ([]byte, error) {

10
pkg/iam/saml_domain_verifier.go
View File

@@ -67,14 +67,18 @@ func NewSAMLDomainVerifier(
func (v *SAMLDomainVerifier) Run(ctx context.Context) error {
v.logger.InfoCtx(ctx, "starting", log.Duration("interval", v.interval))
for {
v.runOnce(ctx)
v.runOnce(ctx)
ticker := time.NewTicker(v.interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
v.logger.InfoCtx(ctx, "shutting down")
return ctx.Err()
case <-time.After(v.interval):
case <-ticker.C:
v.runOnce(ctx)
}
}
}

48
pkg/iam/service.go
View File

@@ -5,6 +5,7 @@ import (
"crypto/rsa"
"crypto/x509"
"fmt"
"sync"
"time"
"github.com/prometheus/client_golang/prometheus"
@@ -21,7 +22,6 @@ import (
"go.probo.inc/probo/pkg/iam/oidc"
"go.probo.inc/probo/pkg/iam/saml"
"go.probo.inc/probo/pkg/iam/scim"
"golang.org/x/sync/errgroup"
)
type (
@@ -175,14 +175,48 @@ func NewService(
}
func (s *Service) Run(ctx context.Context) error {
g, ctx := errgroup.WithContext(ctx)
wg := sync.WaitGroup{}
ctx, cancel := context.WithCancelCause(ctx)
defer cancel(context.Canceled)
g.Go(func() error { return s.SAMLService.Run(ctx) })
g.Go(func() error { return s.OIDCService.Run(ctx) })
g.Go(func() error { return s.samlDomainVerifier.Run(ctx) })
g.Go(func() error { return s.SCIMService.Run(ctx) })
samlCtx, stopSAML := context.WithCancel(context.WithoutCancel(ctx))
wg.Go(func() {
if err := s.SAMLService.Run(samlCtx); err != nil {
cancel(fmt.Errorf("saml service crashed: %w", err))
}
})
return g.Wait()
oidcCtx, stopOIDC := context.WithCancel(context.WithoutCancel(ctx))
wg.Go(func() {
if err := s.OIDCService.Run(oidcCtx); err != nil {
cancel(fmt.Errorf("oidc service crashed: %w", err))
}
})
domainVerifierCtx, stopDomainVerifier := context.WithCancel(context.WithoutCancel(ctx))
wg.Go(func() {
if err := s.samlDomainVerifier.Run(domainVerifierCtx); err != nil {
cancel(fmt.Errorf("saml domain verifier crashed: %w", err))
}
})
scimCtx, stopSCIM := context.WithCancel(context.WithoutCancel(ctx))
wg.Go(func() {
if err := s.SCIMService.Run(scimCtx); err != nil {
cancel(fmt.Errorf("scim service crashed: %w", err))
}
})
<-ctx.Done()
stopSAML()
stopOIDC()
stopDomainVerifier()
stopSCIM()
wg.Wait()
return context.Cause(ctx)
}
func (s *Service) GetMembership(ctx context.Context, membershipID gid.GID) (*coredata.Membership, error) {