127 lines
3.4 KiB
Go
127 lines
3.4 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"time"
|
|
|
|
"github.com/segmentio/kafka-go"
|
|
"github.com/tinode/chat/server/logs"
|
|
"github.com/tinode/chat/server/store"
|
|
"github.com/tinode/chat/server/store/types"
|
|
)
|
|
|
|
const (
|
|
kafkaOutboxPollInterval = 2 * time.Second
|
|
kafkaOutboxTenantLimit = 1000
|
|
kafkaOutboxBatchLimit = 100
|
|
)
|
|
|
|
func startKafkaOutboxPoller() {
|
|
if !kafkaEnabled() {
|
|
return
|
|
}
|
|
kafkaBus.wg.Add(1)
|
|
go func() {
|
|
defer kafkaBus.wg.Done()
|
|
ticker := time.NewTicker(kafkaOutboxPollInterval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-kafkaBus.ctx.Done():
|
|
return
|
|
default:
|
|
drainKafkaOutbox(kafkaBus.ctx)
|
|
}
|
|
select {
|
|
case <-kafkaBus.ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
func enqueueKafkaOutbox(tenantID types.TenantID, topicName, key string, envelope kafkaEnvelope) error {
|
|
if envelope.OccurredAt.IsZero() {
|
|
envelope.OccurredAt = time.Now().UTC()
|
|
}
|
|
raw, err := json.Marshal(envelope)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return store.HTTPInfrastructure.CreateKafkaOutbox(tenantID, &types.KafkaOutboxRecord{
|
|
EventID: envelope.EventID,
|
|
Topic: topicName,
|
|
EventType: envelope.EventType,
|
|
BizType: envelope.BizType,
|
|
BizID: envelope.BizID,
|
|
Key: key,
|
|
Envelope: raw,
|
|
Status: types.KafkaOutboxPending,
|
|
NextRetryAt: nil,
|
|
})
|
|
}
|
|
|
|
func drainKafkaOutbox(ctx context.Context) {
|
|
if store.Tenants == nil || store.HTTPInfrastructure == nil {
|
|
return
|
|
}
|
|
tenants, err := store.Tenants.ListActive(kafkaOutboxTenantLimit)
|
|
if err != nil {
|
|
logs.Warn.Printf("kafka outbox: list active tenants: %v", err)
|
|
return
|
|
}
|
|
now := time.Now().UTC().Round(time.Millisecond)
|
|
for _, tenant := range tenants {
|
|
if tenant.ID.IsZero() {
|
|
continue
|
|
}
|
|
records, err := store.HTTPInfrastructure.ListPendingKafkaOutbox(tenant.ID, kafkaOutboxBatchLimit, now)
|
|
if err != nil {
|
|
logs.Warn.Printf("kafka outbox: list tenant=%d: %v", tenant.ID, err)
|
|
continue
|
|
}
|
|
for _, record := range records {
|
|
if err = publishKafkaOutboxRecord(ctx, record); err != nil {
|
|
nextRetryAt := now.Add(kafkaOutboxRetryDelay(record.Attempts + 1))
|
|
if markErr := store.HTTPInfrastructure.MarkKafkaOutboxFailed(tenant.ID, record.EventID,
|
|
"KAFKA_PUBLISH_FAILED", err.Error(), nextRetryAt, now); markErr != nil {
|
|
logs.Warn.Printf("kafka outbox: mark failed event=%s tenant=%d: %v", record.EventID, tenant.ID, markErr)
|
|
}
|
|
continue
|
|
}
|
|
if err = store.HTTPInfrastructure.MarkKafkaOutboxSent(tenant.ID, record.EventID, time.Now().UTC().Round(time.Millisecond)); err != nil {
|
|
logs.Warn.Printf("kafka outbox: mark sent event=%s tenant=%d: %v", record.EventID, tenant.ID, err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func publishKafkaOutboxRecord(ctx context.Context, record types.KafkaOutboxRecord) error {
|
|
topic := kafkaTopic(record.Topic)
|
|
if topic == "" {
|
|
topic = record.Topic
|
|
}
|
|
return kafkaBus.writer.WriteMessages(ctx, kafka.Message{
|
|
Topic: topic,
|
|
Key: []byte(record.Key),
|
|
Value: record.Envelope,
|
|
Time: time.Now().UTC(),
|
|
})
|
|
}
|
|
|
|
func kafkaOutboxRetryDelay(attempt int) time.Duration {
|
|
if attempt <= 0 {
|
|
attempt = 1
|
|
}
|
|
if kafkaBus == nil || len(kafkaBus.config.RetryBackoffMS) == 0 {
|
|
return time.Duration(attempt) * 10 * time.Second
|
|
}
|
|
index := attempt - 1
|
|
if index >= len(kafkaBus.config.RetryBackoffMS) {
|
|
index = len(kafkaBus.config.RetryBackoffMS) - 1
|
|
}
|
|
return time.Duration(kafkaBus.config.RetryBackoffMS[index]) * time.Millisecond
|
|
}
|