Method Start

workers/entity/controller.go:132–163

()

}

func (c *Controller) Start() error {
	c.mux.Lock()
	defer c.mux.Unlock()

	if c.running {
		return nil
	}
	c.quit = nil

	c.loadAllEnterprises()
	c.loadAllOrganizations()
	c.loadAllRepositories()

	consumer, err := watcher.RegisterConsumer(
		c.ctx, c.consumerID,
		composeControllerWatcherFilters(),
	)
	if err != nil {
		return fmt.Errorf("failed to create consumer for entity controller: %w", err)
	}

	c.consumer = consumer
	c.running = true
	c.quit = make(chan struct{})
	c.eventQueue = workersCommon.NewUnboundedChan[dbCommon.ChangePayload](c.ctx, c.quit)

	go c.loop()
	go c.eventQueue.Process(c.handleWatcherEvent)
	go workersCommon.RetryLoop(c.ctx, c.quit, retryLoopInterval, c.retryFailedWorkers)

	return nil
}

func (c *Controller) Stop() error {
	slog.DebugContext(c.ctx, "stopping entity controller", "entity", c.consumerID)