111 lines
2.7 KiB
Go
111 lines
2.7 KiB
Go
package notify
|
|
|
|
import "context"
|
|
import "sync"
|
|
import "time"
|
|
|
|
import "codit/internal/db"
|
|
import "codit/internal/models"
|
|
import codit_logger "codit/logger"
|
|
|
|
const relayInterval time.Duration = 5 * time.Second
|
|
const relayBatch int = 50
|
|
|
|
// Relay is the single external-delivery path: it drains events not yet dispatched
|
|
// (the transactional outbox) and hands each to the Dispatcher, then stamps it.
|
|
// Every producer that writes an event row gets external alerting for free — no
|
|
// per-producer wiring. In-app feed delivery is independent of this.
|
|
type Relay struct {
|
|
store *db.Store
|
|
dispatcher *Dispatcher
|
|
logger codit_logger.Logger
|
|
stopCh chan struct{}
|
|
stopOnce sync.Once
|
|
wg sync.WaitGroup
|
|
}
|
|
|
|
func NewRelay(store *db.Store, dispatcher *Dispatcher, logger codit_logger.Logger) *Relay {
|
|
return &Relay{
|
|
store: store,
|
|
dispatcher: dispatcher,
|
|
logger: logger,
|
|
stopCh: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
func (r *Relay) log(level codit_logger.LogLevel, format string, args ...any) {
|
|
if r.logger != nil {
|
|
r.logger.Write(logIDNotify, level, format, args...)
|
|
}
|
|
}
|
|
|
|
func (r *Relay) Start(ctx context.Context) {
|
|
r.wg.Add(1)
|
|
go r.loop(ctx)
|
|
}
|
|
|
|
func (r *Relay) Stop() {
|
|
r.stopOnce.Do(func() { close(r.stopCh) })
|
|
}
|
|
|
|
func (r *Relay) Wait() {
|
|
r.wg.Wait()
|
|
}
|
|
|
|
func (r *Relay) loop(ctx context.Context) {
|
|
var ticker *time.Ticker
|
|
defer r.wg.Done()
|
|
ticker = time.NewTicker(relayInterval)
|
|
defer ticker.Stop()
|
|
r.drain(ctx)
|
|
for {
|
|
select {
|
|
case <-r.stopCh:
|
|
return
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
r.drain(ctx)
|
|
}
|
|
}
|
|
}
|
|
|
|
// drain dispatches a batch of pending events. External delivery is at-most-once:
|
|
// an event is stamped dispatched even if delivery failed (failures are logged),
|
|
// so a flaky channel can't wedge the relay — the durable record is the in-app feed.
|
|
func (r *Relay) drain(ctx context.Context) {
|
|
var events []models.Event
|
|
var ev models.Event
|
|
var now int64
|
|
var err error
|
|
|
|
events, err = r.store.ListUndispatchedEvents(relayBatch)
|
|
if err != nil {
|
|
r.log(codit_logger.LOG_ERROR, "relay: list undispatched failed err=%v", err)
|
|
return
|
|
}
|
|
for _, ev = range events {
|
|
select {
|
|
case <-r.stopCh:
|
|
return
|
|
case <-ctx.Done():
|
|
return
|
|
default:
|
|
}
|
|
var dctx context.Context
|
|
var cancel context.CancelFunc
|
|
dctx, cancel = context.WithTimeout(context.Background(), r.store.SettingDuration(db.SettingTimeoutRelayDispatchSec))
|
|
err = r.dispatcher.Notify(dctx, ev)
|
|
cancel()
|
|
if err != nil {
|
|
r.log(codit_logger.LOG_ERROR, "relay: dispatch event=%q kind=%q err=%v", ev.ID, ev.Kind, err)
|
|
}
|
|
now = time.Now().UTC().Unix()
|
|
err = r.store.MarkEventDispatched(context.Background(), ev.ID, now)
|
|
if err != nil {
|
|
r.log(codit_logger.LOG_ERROR, "relay: mark dispatched event=%q err=%v", ev.ID, err)
|
|
return
|
|
}
|
|
}
|
|
}
|