Function NewDBBatcher

enterprise/coderd/connectionlog/connectionlog.go:155–190

NewDBBatcher creates a DBBatcher that batches writes to the database and starts its background processing loop. Close must be called to flush remaining entries on shutdown.

(ctx context.Context, store database.Store, log slog.Logger, opts ...DBBatcherOption)

// and starts its background processing loop. Close must be called to
// flush remaining entries on shutdown.
func NewDBBatcher(ctx context.Context, store database.Store, log slog.Logger, opts ...DBBatcherOption) *DBBatcher {
	b := &DBBatcher{
		store: store,
		log:   log,
		clock: quartz.NewReal(),
	}

	for _, opt := range opts {
		opt(b)
	}

	if b.interval == 0 {
		b.interval = defaultFlushInterval
	}
	if b.maxBatchSize == 0 {
		b.maxBatchSize = defaultBatchSize
	}

	b.timer = b.clock.NewTimer(b.interval)
	b.itemCh = make(chan database.UpsertConnectionLogParams, b.maxBatchSize)
	b.dedupedBatch = make(map[uuid.UUID]batchEntry, b.maxBatchSize)
	b.retryCh = make(chan database.BatchUpsertConnectionLogsParams, retryQueueSize)

	b.ctx, b.cancel = context.WithCancel(ctx)
	b.wg.Add(2)
	go func() {
		defer b.wg.Done()
		b.run(b.ctx)
	}()
	go func() {
		defer b.wg.Done()
		b.retryLoop()
	}()

	return b
}

// Upsert enqueues a connection log entry for batched writing. It
// blocks if the internal buffer is full, ensuring no logs are dropped.