mirror of
https://github.com/pomerium/pomerium.git
synced 2025-07-21 10:38:08 +02:00
a95423b310
* core/identity: add data store for thread-safe storage of sessions and users * wip * add test * wip * clean up context * fix nil session error * add stop message * remove log * use origin context * use base context for manager calls * use manager context for syncers too * add runtime flag * rename legacy lease * add comment * use NotSame * add comment * Update internal/identity/manager/manager.go Co-authored-by: Kenneth Jenkins <51246568+kenjenkins@users.noreply.github.com> * lint --------- Co-authored-by: Kenneth Jenkins <51246568+kenjenkins@users.noreply.github.com>
79 lines
2.3 KiB
Go
79 lines
2.3 KiB
Go
package manager
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/pomerium/pomerium/internal/log"
|
|
"github.com/pomerium/pomerium/pkg/grpc/databroker"
|
|
"github.com/pomerium/pomerium/pkg/grpc/session"
|
|
"github.com/pomerium/pomerium/pkg/grpc/user"
|
|
"github.com/pomerium/pomerium/pkg/grpcutil"
|
|
)
|
|
|
|
type sessionSyncerHandler struct {
|
|
baseCtx context.Context
|
|
mgr *Manager
|
|
}
|
|
|
|
func newSessionSyncer(ctx context.Context, mgr *Manager) *databroker.Syncer {
|
|
return databroker.NewSyncer("identity_manager/sessions", sessionSyncerHandler{baseCtx: ctx, mgr: mgr},
|
|
databroker.WithTypeURL(grpcutil.GetTypeURL(new(session.Session))))
|
|
}
|
|
|
|
func (h sessionSyncerHandler) ClearRecords(ctx context.Context) {
|
|
h.mgr.onDeleteAllSessions(ctx)
|
|
}
|
|
|
|
func (h sessionSyncerHandler) GetDataBrokerServiceClient() databroker.DataBrokerServiceClient {
|
|
return h.mgr.cfg.Load().dataBrokerClient
|
|
}
|
|
|
|
func (h sessionSyncerHandler) UpdateRecords(ctx context.Context, _ uint64, records []*databroker.Record) {
|
|
for _, record := range records {
|
|
if record.GetDeletedAt() != nil {
|
|
h.mgr.onDeleteSession(h.baseCtx, record.GetId())
|
|
} else {
|
|
var s session.Session
|
|
err := record.Data.UnmarshalTo(&s)
|
|
if err != nil {
|
|
log.Ctx(ctx).Warn().Err(err).Msg("invalid data in session record, ignoring")
|
|
} else {
|
|
h.mgr.onUpdateSession(h.baseCtx, &s)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
type userSyncerHandler struct {
|
|
baseCtx context.Context
|
|
mgr *Manager
|
|
}
|
|
|
|
func newUserSyncer(ctx context.Context, mgr *Manager) *databroker.Syncer {
|
|
return databroker.NewSyncer("identity_manager/users", userSyncerHandler{baseCtx: ctx, mgr: mgr},
|
|
databroker.WithTypeURL(grpcutil.GetTypeURL(new(user.User))))
|
|
}
|
|
|
|
func (h userSyncerHandler) ClearRecords(ctx context.Context) {
|
|
h.mgr.onDeleteAllUsers(ctx)
|
|
}
|
|
|
|
func (h userSyncerHandler) GetDataBrokerServiceClient() databroker.DataBrokerServiceClient {
|
|
return h.mgr.cfg.Load().dataBrokerClient
|
|
}
|
|
|
|
func (h userSyncerHandler) UpdateRecords(ctx context.Context, _ uint64, records []*databroker.Record) {
|
|
for _, record := range records {
|
|
if record.GetDeletedAt() != nil {
|
|
h.mgr.onDeleteUser(h.baseCtx, record.GetId())
|
|
} else {
|
|
var u user.User
|
|
err := record.Data.UnmarshalTo(&u)
|
|
if err != nil {
|
|
log.Ctx(ctx).Warn().Err(err).Msg("invalid data in user record, ignoring")
|
|
} else {
|
|
h.mgr.onUpdateUser(h.baseCtx, &u)
|
|
}
|
|
}
|
|
}
|
|
}
|