refactor(background): parameterize sync queue and worker pool constructors

Split each constructor into a default-args wrapper and a config-accepting
variant so the sync queue interval/batch size and the worker pool size/
queue cap can be sourced from the settings registry at startup. These
values are constructed once at boot, so they are tagged requires_restart
in the admin UI.

queue.go:
- NewSyncQueueProcessorWithConfig(db, interval, batchSize) takes the
  flush interval and batch size as parameters; NewSyncQueueProcessor
  becomes a thin wrapper with the historical 5s / 50 defaults.

worker.go:
- NewWorkerWithConfig(numWorkers, queueCap, connManager) takes the
  queue capacity as a parameter; NewWorker becomes a thin wrapper with
  the historical cap of 100.

No behavior change for existing callers; main.go will switch to the
config-accepting variants in a follow-up wiring commit.
This commit is contained in:
2026-08-10 08:02:02 -04:00
parent 757398bf15
commit 936a48405b
2 changed files with 18 additions and 3 deletions
+9 -1
View File
@@ -176,10 +176,17 @@ func (w *Worker) GetActiveJobCount() int {
return count
}
func NewWorker(numWorkers int, connManager *wsync.ConnectionManager) *Worker {
return NewWorkerWithConfig(numWorkers, 100, connManager)
}
// NewWorkerWithConfig constructs a worker pool with the given worker count and
// job-queue capacity. Used at startup to source values from the settings
// registry.
func NewWorkerWithConfig(numWorkers, queueCap int, connManager *wsync.ConnectionManager) *Worker {
ctx, cancel := context.WithCancel(context.Background())
w := &Worker{
jobQueue: make(chan *Job, 100),
jobQueue: make(chan *Job, queueCap),
results: make(map[string]*JobResult),
ctx: ctx,
cancel: cancel,
@@ -995,3 +1002,4 @@ func (w *Worker) Shutdown() {
close(w.jobQueue)
w.wg.Wait()
}
+9 -2
View File
@@ -74,11 +74,18 @@ type SyncQueueItem struct {
}
func NewSyncQueueProcessor(db *database.Queries) *SyncQueueProcessor {
return NewSyncQueueProcessorWithConfig(db, 5*time.Second, 50)
}
// NewSyncQueueProcessorWithConfig constructs a processor with the given flush
// interval and batch size. Used at startup to source values from the settings
// registry.
func NewSyncQueueProcessorWithConfig(db *database.Queries, interval time.Duration, batchSize int) *SyncQueueProcessor {
return &SyncQueueProcessor{
db: db,
progressChan: make(chan *ProgressUpdate, 100),
interval: 5 * time.Second,
batchSize: 50,
interval: interval,
batchSize: batchSize,
}
}