From 3ffdcb7263a24f418cb69574cf230fdfb1f7731e Mon Sep 17 00:00:00 2001 From: suifri Date: Thu, 10 Sep 2026 12:02:02 +0300 Subject: [PATCH] [WTEL-10383]feature(broker): add creating system online skill on new domain event - add creation of simple topic broker queue - add queue configuration object - add new domain created handler --- app/app.go | 1 + app/online_skills.go | 15 ++ model/domain.go | 40 ++++- mq/layered_mq.go | 6 +- mq/mq.go | 4 + mq/rabbit/broker_queue.go | 294 ++++++++++++++++++++++++++++++++ mq/rabbit/client.go | 88 ++++++++++ mq/rabbit/config.go | 121 +++++++++++++ store/sqlstore/online_skills.go | 29 ++++ store/store.go | 1 + 10 files changed, 597 insertions(+), 2 deletions(-) create mode 100644 mq/rabbit/broker_queue.go create mode 100644 mq/rabbit/config.go diff --git a/app/app.go b/app/app.go index 10c9ec30..8b1329f6 100644 --- a/app/app.go +++ b/app/app.go @@ -200,6 +200,7 @@ func New(options ...string) (outApp *App, outErr error) { app.Store = store.NewLayeredStore(sqlSupplier) app.MessageQueue = rabbit.NewRabbitMQ(app.Config().NodeName, &app.Config().MessageQueueSettings) + app.initDomainEventListener() app.MessageQueue.Start() app.Hubs = NewHubs(app) diff --git a/app/online_skills.go b/app/online_skills.go index 0a7a22ad..df3c4b22 100644 --- a/app/online_skills.go +++ b/app/online_skills.go @@ -4,8 +4,13 @@ import ( "context" "github.com/webitel/engine/model" + "github.com/webitel/wlog" ) +func (app *App) initDomainEventListener() { + app.MessageQueue.SetDomainsEventHandler(app.handleDomainEventCreated) +} + func (app *App) CreateOnlineSkills(ctx context.Context, preset *model.OnlineSkills) (*model.OnlineSkills, model.AppError) { preset.PreSave() @@ -35,3 +40,13 @@ func (app *App) PatchOnlineSkills(ctx context.Context, cmd *model.PatchOnlineSki func (app *App) DeleteOnlineSkills(ctx context.Context, cmd *model.DeleteSkillPresetCmd) model.AppError { return app.Store.OnlineSkills().Delete(ctx, cmd) } + +func (app *App) handleDomainEventCreated(ctx context.Context, e *model.DomainEvent) error { + if err := app.Store.OnlineSkills().CreateSystem(ctx, e.ID); err != nil { + app.Log.Error("processing domain created event", wlog.Err(err)) + + return err + } + + return nil +} diff --git a/model/domain.go b/model/domain.go index d4081c24..44ad13d1 100644 --- a/model/domain.go +++ b/model/domain.go @@ -1,6 +1,44 @@ package model -import "time" +import ( + "fmt" + "strconv" + "strings" + "time" +) + +type DomainEvent struct { + ID int64 +} + +func NewDomainEventFromRoutingKey(rk string) (*DomainEvent, AppError) { + splitted := strings.Split(rk, ".") + if len(splitted) < 3 { + return nil, NewBadRequestError( + "model.domain.new_domain_event.invalid_rk_len", + "received roting key with len less than 3", + ) + } + + const domainIdIndex = 2 + + parsedDomainId, err := strconv.ParseInt(splitted[domainIdIndex], 10, 64) + if err != nil { + return nil, NewBadRequestError( + "model.domain.new_domain_event.parsing_id", + fmt.Sprintf("parsing routing key id to integer: %+v", err), + ) + } + + if parsedDomainId <= 0 { + return nil, NewBadRequestError( + "model.domain.new_domain_event.domain_id_less_or_equal_zero", + fmt.Sprintf("received domain id less or equal zero: %d", parsedDomainId), + ) + } + + return &DomainEvent{ID: parsedDomainId}, nil +} type DomainProvider interface { Domain(int64) int64 diff --git a/mq/layered_mq.go b/mq/layered_mq.go index 4d07e4e9..fd459760 100644 --- a/mq/layered_mq.go +++ b/mq/layered_mq.go @@ -70,6 +70,10 @@ func (l *LayeredMQ) Send(ctx context.Context, exchange string, rk string, body [ return l.MQLayer.Send(ctx, exchange, rk, body) } -func (l *LayeredMQ) SendStartFlow(ctx context.Context, domainId int64, schemaId int32, in interface{}) model.AppError { +func (l *LayeredMQ) SendStartFlow(ctx context.Context, domainId int64, schemaId int32, in any) model.AppError { return l.MQLayer.SendStartFlow(ctx, domainId, schemaId, in) } + +func (l *LayeredMQ) SetDomainsEventHandler(h DomainEventHandler) { + l.MQLayer.SetDomainsEventHandler(h) +} diff --git a/mq/mq.go b/mq/mq.go index 7bb6d27c..bc762395 100644 --- a/mq/mq.go +++ b/mq/mq.go @@ -2,9 +2,12 @@ package mq import ( "context" + "github.com/webitel/engine/model" ) +type DomainEventHandler func(ctx context.Context, e *model.DomainEvent) error + type MQ interface { SendJSON(name string, data []byte) model.AppError BindCallEvents(domainId, userId int64) error @@ -25,6 +28,7 @@ type MQ interface { Send(ctx context.Context, exchange string, rk string, body []byte) error SendStartFlow(ctx context.Context, domainId int64, schemaId int32, in interface{}) model.AppError + SetDomainsEventHandler(h DomainEventHandler) } type DomainQueue interface { diff --git a/mq/rabbit/broker_queue.go b/mq/rabbit/broker_queue.go new file mode 100644 index 00000000..57ce614d --- /dev/null +++ b/mq/rabbit/broker_queue.go @@ -0,0 +1,294 @@ +package rabbit + +import ( + "cmp" + "context" + "fmt" + "sync" + "time" + + "github.com/pkg/errors" + amqp "github.com/rabbitmq/amqp091-go" + "github.com/webitel/engine/model" + "github.com/webitel/wlog" +) + +type ( + BrokerQueueOption func(*BrokerQueue) + ChannelProvider func() (*amqp.Channel, error) + Handler func(ctx context.Context, d amqp.Delivery) error +) + +func WithConsumerTag(tag string) BrokerQueueOption { + return func(bq *BrokerQueue) { + bq.consumerTag = tag + } +} + +func WithConcurrency(n int) BrokerQueueOption { + return func(bq *BrokerQueue) { + bq.concurrency = n + } +} + +type BrokerQueue struct { + queueCfg *QueueConfig + bindings []*QueueBindConfig + getChannel ChannelProvider + + consumerTag string + concurrency int + reconnectSec int + + log *wlog.Logger + mx sync.Mutex + channel *amqp.Channel + stop chan struct{} + stopped chan struct{} + handler Handler +} + +func NewBrokerQueue(cfg *QueueConfig, getChannel ChannelProvider, handler Handler, opts ...BrokerQueueOption) *BrokerQueue { + b := &BrokerQueue{ + queueCfg: cfg, + getChannel: getChannel, + concurrency: 1, + reconnectSec: RECONNECT_SEC, + log: wlog.GlobalLogger().With( + wlog.Namespace("context"), + wlog.String("protocol", "amqp"), + wlog.String("queue", cfg.Name), + ), + stop: make(chan struct{}), + stopped: make(chan struct{}), + handler: handler, + } + + for _, o := range opts { + o(b) + } + + return b +} + +func (b *BrokerQueue) Bind(cfg *QueueBindConfig) *BrokerQueue { + b.bindings = append(b.bindings, cfg) + + return b +} + +func (b *BrokerQueue) setup(ch *amqp.Channel) model.AppError { + _, err := ch.QueueDeclare( + b.queueCfg.Name, + b.queueCfg.Durable, + b.queueCfg.AutoDelete, + b.queueCfg.Exclusive, + b.queueCfg.NoWait, + amqp.Table(b.queueCfg.Args), + ) + + if err != nil { + return model.NewCustomCodeError( + "rabbit.broker.queue.setup.declare", + fmt.Sprintf("declaring new queue: %+v", err), + 500, + ) + } + + for _, bind := range b.bindings { + name := cmp.Or(bind.Name, b.queueCfg.Name) + if err := ch.QueueBind(name, bind.Key, bind.Exchange, bind.NoWait, amqp.Table(bind.Args)); err != nil { + return model.NewCustomCodeError( + "rabbit.broker.queue.setup.bind", + fmt.Sprintf("bind %q -> %q (%q): %+v", name, bind.Exchange, bind.Key, err), + 500, + ) + } + } + + return nil +} + +func (b *BrokerQueue) Run(ctx context.Context) model.AppError { + defer close(b.stopped) + + for { + select { + case <-b.stop: + return nil + case <-ctx.Done(): + return model.NewCustomCodeError( + "rabbit.broker.queue.run_context_done", + fmt.Sprintf("context done on running broker queue %s: %+v", b.queueCfg.Name, ctx.Err()), + 408, + ) + default: + } + + ch, err := b.getChannel() + if err != nil { + b.log.Error("getting AMQP channel", wlog.Err(err)) + if !b.sleep(ctx) { + return nil + } + + continue + } + + if err := b.setup(ch); err != nil { + b.log.Error("setup queue", wlog.Err(err)) + b.closeChan(ch) + if !b.sleep(ctx) { + return nil + } + + continue + } + + if err := ch.Qos(b.concurrency, 0, false); err != nil { + b.log.Error("qos", wlog.Err(err)) + } + + deliveries, err := ch.Consume( + b.queueCfg.Name, + b.consumerTag, + false, + b.queueCfg.Exclusive, + false, + b.queueCfg.NoWait, + nil, + ) + + if err != nil { + b.log.Error("consuming queue deliveries", wlog.Err(err)) + b.closeChan(ch) + if !b.sleep(ctx) { + return nil + } + + continue + } + + b.mx.Lock() + b.channel = ch + b.mx.Unlock() + + closeErr := make(chan *amqp.Error, 1) + ch.NotifyClose(closeErr) + + b.log.Info("consuming started") + + cont := b.consumeLoop(ctx, deliveries, closeErr) + b.closeChan(ch) + + if !cont { + return nil + } + + b.log.Warn("channel lost, reconnecting") + } +} + +func (b *BrokerQueue) closeChan(ch *amqp.Channel) { + if ch == nil { + return + } + + if err := ch.Close(); err != nil && !errors.Is(err, amqp.ErrClosed) { + b.log.Warn("closing amqp channel", wlog.Err(err)) + } +} + +func (b *BrokerQueue) process(ctx context.Context, d amqp.Delivery) { + defer func() { + if r := recover(); r != nil { + b.log.Error(fmt.Sprintf("panic in handler: %v", r)) + _ = d.Nack(false, false) + } + }() + + if err := b.handler(ctx, d); err != nil { + b.log.Error("handler error", wlog.Err(err)) + + _ = d.Nack(false, false) + + return + } + + _ = d.Ack(false) +} + +func (b *BrokerQueue) consumeLoop(ctx context.Context, deliveries <-chan amqp.Delivery, closeErr <-chan *amqp.Error) bool { + sem := make(chan struct{}, b.concurrency) + + var wg sync.WaitGroup + + for { + select { + case <-b.stop: + wg.Wait() + return false + case <-ctx.Done(): + wg.Wait() + return false + case <-closeErr: + wg.Wait() + return true + case d, ok := <-deliveries: + if !ok { + wg.Wait() + return true + } + + select { + case sem <- struct{}{}: + case <-ctx.Done(): + wg.Wait() + return false + case <-b.stop: + wg.Wait() + return false + case <-closeErr: + wg.Wait() + return true + } + + wg.Add(1) + go func(d amqp.Delivery) { + defer wg.Done() + defer func() { <-sem }() + + b.process(ctx, d) + }(d) + } + } +} + +func (b *BrokerQueue) sleep(ctx context.Context) bool { + select { + case <-ctx.Done(): + return false + case <-time.After(time.Duration(b.reconnectSec) * time.Second): + return true + case <-b.stop: + return false + } +} + +func (b *BrokerQueue) Close() { + close(b.stop) + + select { + case <-b.stopped: + case <-time.After(time.Duration(5) * time.Second): + b.log.Warn("close timed out waiting for consume loop to stop") + } + + b.mx.Lock() + defer b.mx.Unlock() + + if b.channel != nil && !b.channel.IsClosed() { + _ = b.channel.Cancel(b.consumerTag, false) + _ = b.channel.Close() + } +} diff --git a/mq/rabbit/client.go b/mq/rabbit/client.go index 4d8cb2d2..c0e74499 100644 --- a/mq/rabbit/client.go +++ b/mq/rabbit/client.go @@ -57,6 +57,7 @@ type AMQP struct { registerDomainQueue chan mq.DomainQueue unRegisterDomainQueue chan mq.DomainQueue log *wlog.Logger + domainEventHandler mq.DomainEventHandler mx sync.Mutex } @@ -92,6 +93,12 @@ func (a *AMQP) NewDomainQueue(domainId int64, bindings model.GetAllBindings) (mq func (a *AMQP) Start() { a.initConnection() + + if err := a.initDomains(context.Background()); err != nil { + a.log.Critical("initializing domains consumer", wlog.Err(err)) + panic(err) + } + go a.Listen() } @@ -111,6 +118,24 @@ func (a *AMQP) Ping(context.Context) error { return nil } +func (a *AMQP) BrokerChannelProvider() ChannelProvider { + return func() (*amqp.Channel, error) { + a.mx.Lock() + conn := a.connection + a.mx.Unlock() + + if conn == nil || conn.IsClosed() { + return nil, model.NewCustomCodeError( + "rabbit.client.broker_channel_provider", + "connection closed", + 412, + ) + } + + return conn.Channel() + } +} + func (a *AMQP) addDomainQueue(id int64, q mq.DomainQueue) { a.mx.Lock() defer a.mx.Unlock() @@ -308,3 +333,66 @@ func (a *AMQP) SendStartFlow(ctx context.Context, domainId int64, schemaId int32 return nil } + +func (a *AMQP) initDomains(ctx context.Context) error { + queueCfg := NewQueueConfig( + "engine.domains.consumer", + WithQueueDurable(true), + WithQueueArg("x-queue-type", "quorum"), + ) + + bq := NewBrokerQueue( + queueCfg, + a.BrokerChannelProvider(), + a.processDomainDeliveries, + WithConsumerTag( + fmt.Sprintf("engine-%s-domains", a.nodeName), + ), + WithConcurrency(2), + ) + + bq.Bind(NewQueueBindConfig( + queueCfg.Name, + "domains.create.*", + "webitel", + WithQueueBindNoWait(false), + )) + + go func() { + if err := bq.Run(ctx); err != nil { + a.log.Error("broker queue stopped", wlog.Err(err)) + } + }() + + go func() { + <-a.stopped + + bq.Close() + }() + + return nil +} + +func (a *AMQP) processDomainDeliveries(ctx context.Context, d amqp.Delivery) error { + de, err := model.NewDomainEventFromRoutingKey(d.RoutingKey) + if err != nil { + return err + } + + a.mx.Lock() + h := a.domainEventHandler + a.mx.Unlock() + + if h == nil { + a.log.Warn("no domain event handler set, dropping event") + return nil + } + + return h(ctx, de) +} + +func (a *AMQP) SetDomainsEventHandler(h mq.DomainEventHandler) { + a.mx.Lock() + a.domainEventHandler = h + a.mx.Unlock() +} diff --git a/mq/rabbit/config.go b/mq/rabbit/config.go new file mode 100644 index 00000000..f7da6345 --- /dev/null +++ b/mq/rabbit/config.go @@ -0,0 +1,121 @@ +package rabbit + +type Table map[string]any + +type QueueConfig struct { + Name string + Durable bool + AutoDelete bool + Exclusive bool + NoWait bool + Args Table +} + +func NewQueueConfig(name string, opts ...QueueConfigOption) *QueueConfig { + cfg := &QueueConfig{ + Name: name, + Durable: false, + AutoDelete: false, + Exclusive: false, + NoWait: true, + Args: make(Table), + } + + for _, o := range opts { + o(cfg) + } + + return cfg +} + +type QueueConfigOption func(*QueueConfig) + +func WithQueueName(name string) QueueConfigOption { + return func(qc *QueueConfig) { + qc.Name = name + } +} + +func WithQueueDurable(d bool) QueueConfigOption { + return func(qc *QueueConfig) { + qc.Durable = d + } +} + +func WithQueueAutoDelete(ad bool) QueueConfigOption { + return func(qc *QueueConfig) { + qc.AutoDelete = ad + } +} + +func WithQueueExclusive(e bool) QueueConfigOption { + return func(qc *QueueConfig) { + qc.Exclusive = e + } +} + +func WithQueueNoWait(w bool) QueueConfigOption { + return func(qc *QueueConfig) { + qc.NoWait = w + } +} + +func WithQueueArg(key string, value any) QueueConfigOption { + return func(qc *QueueConfig) { + if qc.Args == nil { + return + } + + qc.Args[key] = value + } +} + +type QueueBindConfig struct { + Name string + Key string + Exchange string + NoWait bool + Args Table +} + +func NewQueueBindConfig(name, key, exchange string, opts ...QueueBindConfigOption) *QueueBindConfig { + cfg := &QueueBindConfig{ + Name: name, + Key: key, + Exchange: exchange, + NoWait: true, + Args: make(Table), + } + + for _, o := range opts { + o(cfg) + } + + return cfg +} + +type QueueBindConfigOption func(*QueueBindConfig) + +func WithQueueBindName(name string) QueueBindConfigOption { + return func(qbc *QueueBindConfig) { + qbc.Name = name + } +} + +func WithQueueBindKey(key string) QueueBindConfigOption { + return func(qbc *QueueBindConfig) { + qbc.Key = key + } +} + +func WithQueueBindExchange(exchange string) QueueBindConfigOption { + return func(qbc *QueueBindConfig) { + qbc.Exchange = exchange + } +} + +func WithQueueBindNoWait(noWait bool) QueueBindConfigOption { + return func(qbc *QueueBindConfig) { + qbc.NoWait = noWait + } +} diff --git a/store/sqlstore/online_skills.go b/store/sqlstore/online_skills.go index e1fdf4d7..515acca0 100644 --- a/store/sqlstore/online_skills.go +++ b/store/sqlstore/online_skills.go @@ -295,3 +295,32 @@ func (s *SqlOnlineSkillsStore) Get(ctx context.Context, search *model.GetSkillPr return result, nil } + +func (s *SqlOnlineSkillsStore) CreateSystem(ctx context.Context, domainID int64) model.AppError { + if _, err := s.GetMaster().WithContext(ctx).Exec( + `insert into call_center.cc_online_skills ( + domain_id, created_at, updated_at, name, is_system + ) + select + :DomainID, + now(), + now(), + :StandartSkill, + true + where not exists ( + select 1 + from call_center.cc_online_skills + where domain_id = :DomainID + and is_system is true + ); + `, + map[string]any{ + "DomainID": domainID, + "StandartSkill": model.StandartOnlineSkill, + }, + ); err != nil { + return model.NewCustomCodeError("sqlstore.online_skills.create_system", err.Error(), extractCodeFromErr(err)) + } + + return nil +} diff --git a/store/store.go b/store/store.go index 59a649e3..dadeb332 100644 --- a/store/store.go +++ b/store/store.go @@ -570,6 +570,7 @@ type OnlineSkillsStore interface { Delete(ctx context.Context, deleteCmd *model.DeleteSkillPresetCmd) model.AppError Search(ctx context.Context, search *model.SearchOnlineSkillsQuery) ([]*model.OnlineSkills, model.AppError) Get(ctx context.Context, search *model.GetSkillPresetQuery) (*model.OnlineSkills, model.AppError) + CreateSystem(ctx context.Context, domainID int64) model.AppError } // ApplyFiltersToBuilder determines type of {filters} parameter and applies {filters} to the {base} according to the determined type.