reindex

Documentation

Index

• Variables
• func RawToSource(raw *model.SearchUserRaw) *model.SearchUserSource
• type ElasticsearchReindexer
• type PostgresqlReindexer
• type ReindexItem
• type ReindexRequest
• type ReindexResult
• type Reindexer
  • func (s *Reindexer) EnqueueReindexUserTask(ctx context.Context, userID string) error
  • func (s *Reindexer) ExecReindexUser(ctx context.Context, request ReindexRequest) (result ReindexResult)
  • func (s *Reindexer) MarkUsersAsReindexRequiredInTx(ctx context.Context, userIDs []string) error
• type ReindexerLogger
  • func NewReindexerLogger(lf *log.Factory) *ReindexerLogger
• type Sink
  • func (s *Sink) ReceiveBlockingEvent(ctx context.Context, e *event.Event) error
  • func (s *Sink) ReceiveNonBlockingEvent(ctx context.Context, e *event.Event) error
• type SinkLogger
  • func NewSinkLogger(lf *log.Factory) SinkLogger
• type SourceProvider
  • func (s *SourceProvider) QueryPage(ctx context.Context, after model.PageCursor, first uint64) ([]ReindexItem, error)
• type UserQueries
• type UserReindexCreateProducer

Variables

var DependencySet = wire.NewSet(
	wire.Struct(new(SourceProvider), "*"),
	NewReindexerLogger,
	wire.Struct(new(Reindexer), "*"),
	NewSinkLogger,
	wire.Struct(new(Sink), "*"),
)

Functions

func RawToSource

func RawToSource(raw *model.SearchUserRaw) *model.SearchUserSource

Types

type ElasticsearchReindexer

type ElasticsearchReindexer interface {
	ReindexUser(user *model.SearchUserSource) error
	DeleteUser(userID string) error
}

type PostgresqlReindexer

type PostgresqlReindexer interface {
	ReindexUser(ctx context.Context, user *model.SearchUserSource) error
	DeleteUser(ctx context.Context, userID string) error
}

type ReindexItem

type ReindexItem struct {
	Value  *model.SearchUserSource
	Cursor model.PageCursor
}

type ReindexRequest

type ReindexRequest struct {
	UserID string `json:"user_id"`
}

type ReindexResult

type ReindexResult struct {
	UserID       string `json:"user_id"`
	IsSuccess    bool   `json:"is_success"`
	ErrorMessage string `json:"error_message,omitempty"`
}

type Reindexer

type Reindexer struct {
	AppID        config.AppID
	SearchConfig *config.SearchConfig
	Clock        clock.Clock
	Database     *appdb.Handle
	Logger       *ReindexerLogger
	UserStore    *user.Store
	Producer     UserReindexCreateProducer

	SourceProvider *SourceProvider

	ElasticsearchReindexer ElasticsearchReindexer
	PostgresqlReindexer    PostgresqlReindexer
}

func (*Reindexer) EnqueueReindexUserTask

func (s *Reindexer) EnqueueReindexUserTask(ctx context.Context, userID string) error

func (*Reindexer) ExecReindexUser

func (s *Reindexer) ExecReindexUser(ctx context.Context, request ReindexRequest) (result ReindexResult)

func (*Reindexer) MarkUsersAsReindexRequiredInTx

func (s *Reindexer) MarkUsersAsReindexRequiredInTx(ctx context.Context, userIDs []string) error

type ReindexerLogger

type ReindexerLogger struct{ *log.Logger }

func NewReindexerLogger

func NewReindexerLogger(lf *log.Factory) *ReindexerLogger

type Sink

type Sink struct {
	Logger    SinkLogger
	Reindexer *Reindexer
	Database  *appdb.Handle
}

func (*Sink) ReceiveBlockingEvent

func (s *Sink) ReceiveBlockingEvent(ctx context.Context, e *event.Event) error

func (*Sink) ReceiveNonBlockingEvent

func (s *Sink) ReceiveNonBlockingEvent(ctx context.Context, e *event.Event) error

type SinkLogger

type SinkLogger struct{ *log.Logger }

func NewSinkLogger

func NewSinkLogger(lf *log.Factory) SinkLogger

type SourceProvider

type SourceProvider struct {
	AppID           config.AppID
	Users           UserQueries
	UserStore       *user.Store
	IdentityService *identityservice.Service
	RolesGroups     *rolesgroups.Store
}

func (*SourceProvider) QueryPage

func (s *SourceProvider) QueryPage(ctx context.Context, after model.PageCursor, first uint64) ([]ReindexItem, error)

type UserQueries

type UserQueries interface {
	Get(ctx context.Context, userID string, role accesscontrol.Role) (*model.User, error)
}

type UserReindexCreateProducer

type UserReindexCreateProducer interface {
	NewTask(appID string, input json.RawMessage, taskIDPrefix string) *redisqueue.Task
	EnqueueTask(ctx context.Context, task *redisqueue.Task) error
}