CUB-117: Port Moonraker + MQTT printer integrations to Go
All checks were successful
Dev Build / build-test (pull_request) Successful in 1m29s
All checks were successful
Dev Build / build-test (pull_request) Successful in 1m29s
- Moonraker REST client with GetPrinterInfo, GetPrintStats, GetPrintHistory - Moonraker WebSocket client with auto-reconnect + telemetry parsing - MQTT client via paho.mqtt.golang with TLS support for Bambu Lab - Moonraker poller worker: background polling, dedup, usage logging to PostgreSQL - MQTT subscriber worker: Bambu telemetry parsing, print job tracking - Config: 7 new env vars (MOONRAKER_URL, MQTT_BROKER, etc.) - main.go: per-printer worker discovery, graceful shutdown
This commit is contained in:
255
backend/internal/workers/moonraker_poller.go
Normal file
255
backend/internal/workers/moonraker_poller.go
Normal file
@@ -0,0 +1,255 @@
|
||||
package workers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/CubeCraft-Creations/Extrudex/backend/internal/clients"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// ── deduplication ───────────────────────────────────────────────────────────
|
||||
|
||||
// jobTrack holds the last-seen filename and filament_used for dedup.
|
||||
type jobTrack struct {
|
||||
filename string
|
||||
filamentUsed float64
|
||||
}
|
||||
|
||||
// MoonrakerPoller periodically queries the Moonraker REST API for print stats
|
||||
// and logs filament usage to PostgreSQL. It deduplicates by tracking the
|
||||
// last-known filament_used value for the active job on this printer.
|
||||
type MoonrakerPoller struct {
|
||||
client *clients.MoonrakerClient
|
||||
pool *pgxpool.Pool
|
||||
pollInterval time.Duration
|
||||
printerID int
|
||||
printerName string
|
||||
|
||||
mu sync.Mutex
|
||||
track jobTrack
|
||||
}
|
||||
|
||||
// MoonrakerPollerConfig holds configuration for the Moonraker polling worker.
|
||||
type MoonrakerPollerConfig struct {
|
||||
Client *clients.MoonrakerClient
|
||||
Pool *pgxpool.Pool
|
||||
PollInterval time.Duration
|
||||
PrinterID int
|
||||
PrinterName string
|
||||
}
|
||||
|
||||
// NewMoonrakerPoller creates a new MoonrakerPoller worker.
|
||||
func NewMoonrakerPoller(cfg MoonrakerPollerConfig) *MoonrakerPoller {
|
||||
if cfg.PollInterval <= 0 {
|
||||
cfg.PollInterval = 10 * time.Second
|
||||
}
|
||||
return &MoonrakerPoller{
|
||||
client: cfg.Client,
|
||||
pool: cfg.Pool,
|
||||
pollInterval: cfg.PollInterval,
|
||||
printerID: cfg.PrinterID,
|
||||
printerName: cfg.PrinterName,
|
||||
}
|
||||
}
|
||||
|
||||
// Run starts the polling loop. It blocks until ctx is cancelled.
|
||||
func (w *MoonrakerPoller) Run(ctx context.Context) {
|
||||
slog.Info("moonraker poller: starting",
|
||||
"printer_id", w.printerID,
|
||||
"printer_name", w.printerName,
|
||||
"interval", w.pollInterval,
|
||||
)
|
||||
|
||||
ticker := time.NewTicker(w.pollInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
w.poll(ctx)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
slog.Info("moonraker poller: stopping", "printer_id", w.printerID)
|
||||
return
|
||||
case <-ticker.C:
|
||||
w.poll(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (w *MoonrakerPoller) poll(ctx context.Context) {
|
||||
stats, err := w.client.GetPrintStats(ctx)
|
||||
if err != nil {
|
||||
slog.Warn("moonraker poller: failed to get print stats",
|
||||
"printer_id", w.printerID, "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
if stats.State == "" {
|
||||
return
|
||||
}
|
||||
|
||||
jobName := "unknown"
|
||||
if stats.Filename != nil {
|
||||
jobName = *stats.Filename
|
||||
}
|
||||
|
||||
// Compute delta under lock; release before I/O.
|
||||
w.mu.Lock()
|
||||
prevName := w.track.filename
|
||||
prevUsed := w.track.filamentUsed
|
||||
|
||||
if jobName != prevName {
|
||||
w.track.filename = jobName
|
||||
w.track.filamentUsed = 0
|
||||
prevUsed = 0
|
||||
}
|
||||
|
||||
deltaMM := stats.FilamentUsedMm - prevUsed
|
||||
totalMM := stats.FilamentUsedMm
|
||||
if deltaMM <= 0 && jobName == prevName {
|
||||
w.mu.Unlock()
|
||||
return
|
||||
}
|
||||
w.mu.Unlock()
|
||||
|
||||
slog.Info("moonraker poller: filament usage",
|
||||
"printer_id", w.printerID,
|
||||
"job", jobName,
|
||||
"delta_mm", deltaMM,
|
||||
"total_mm", totalMM,
|
||||
"state", stats.State,
|
||||
)
|
||||
|
||||
jobID, err := w.ensurePrintJob(ctx, jobName, stats.State)
|
||||
if err != nil {
|
||||
slog.Error("moonraker poller: failed to ensure print job",
|
||||
"printer_id", w.printerID, "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
spoolID, density := lookupActiveSpool(ctx, w.pool, w.printerID)
|
||||
|
||||
if err := insertUsageLog(ctx, w.pool, jobID, spoolID, deltaMM, density); err != nil {
|
||||
slog.Error("moonraker poller: failed to log usage",
|
||||
"printer_id", w.printerID, "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
w.mu.Lock()
|
||||
w.track.filamentUsed = totalMM
|
||||
w.mu.Unlock()
|
||||
}
|
||||
|
||||
func (w *MoonrakerPoller) ensurePrintJob(ctx context.Context, jobName, state string) (int, error) {
|
||||
var jobID int
|
||||
err := w.pool.QueryRow(ctx, `
|
||||
SELECT pj.id FROM print_jobs pj
|
||||
JOIN job_statuses js ON pj.job_status_id = js.id
|
||||
WHERE pj.printer_id = $1
|
||||
AND pj.job_name = $2
|
||||
AND pj.deleted_at IS NULL
|
||||
AND js.name IN ('printing', 'pending')
|
||||
ORDER BY pj.created_at DESC
|
||||
LIMIT 1
|
||||
`, w.printerID, jobName).Scan(&jobID)
|
||||
|
||||
if err == nil {
|
||||
_, _ = w.pool.Exec(ctx, `
|
||||
UPDATE print_jobs SET
|
||||
job_status_id = (SELECT id FROM job_statuses WHERE name = 'printing'),
|
||||
started_at = COALESCE(started_at, NOW()),
|
||||
updated_at = NOW()
|
||||
WHERE id = $1
|
||||
AND job_status_id = (SELECT id FROM job_statuses WHERE name = 'pending')
|
||||
`, jobID)
|
||||
return jobID, nil
|
||||
}
|
||||
|
||||
var statusID int
|
||||
err = w.pool.QueryRow(ctx, `SELECT id FROM job_statuses WHERE name = 'printing'`).Scan(&statusID)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("moonraker poller: missing 'printing' job status: %w", err)
|
||||
}
|
||||
|
||||
err = w.pool.QueryRow(ctx, `
|
||||
INSERT INTO print_jobs (printer_id, job_name, file_name, job_status_id, started_at)
|
||||
VALUES ($1, $2, $3, $4, NOW())
|
||||
RETURNING id
|
||||
`, w.printerID, jobName, jobName, statusID).Scan(&jobID)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("moonraker poller: failed to create print job: %w", err)
|
||||
}
|
||||
|
||||
slog.Info("moonraker poller: created print job", "job_id", jobID, "job_name", jobName)
|
||||
return jobID, nil
|
||||
}
|
||||
|
||||
// ── Package-level helpers (shared by both workers) ──────────────────────────
|
||||
|
||||
// lookupActiveSpool finds the most recently used spool for a given printer.
|
||||
func lookupActiveSpool(ctx context.Context, pool *pgxpool.Pool, printerID int) (int, float64) {
|
||||
type result struct {
|
||||
id int
|
||||
density float64
|
||||
}
|
||||
var res result
|
||||
|
||||
err := pool.QueryRow(ctx, `
|
||||
SELECT fs.id, COALESCE(mb.density_g_cm3, 1.24)
|
||||
FROM filament_spools fs
|
||||
JOIN material_bases mb ON fs.material_base_id = mb.id
|
||||
JOIN print_jobs pj ON pj.filament_spool_id = fs.id
|
||||
WHERE pj.printer_id = $1 AND fs.deleted_at IS NULL
|
||||
ORDER BY pj.created_at DESC LIMIT 1
|
||||
`, printerID).Scan(&res.id, &res.density)
|
||||
if err == nil {
|
||||
return res.id, res.density
|
||||
}
|
||||
|
||||
err = pool.QueryRow(ctx, `
|
||||
SELECT fs.id, COALESCE(mb.density_g_cm3, 1.24)
|
||||
FROM filament_spools fs
|
||||
JOIN material_bases mb ON fs.material_base_id = mb.id
|
||||
WHERE fs.deleted_at IS NULL
|
||||
ORDER BY fs.created_at DESC LIMIT 1
|
||||
`).Scan(&res.id, &res.density)
|
||||
if err == nil {
|
||||
return res.id, res.density
|
||||
}
|
||||
|
||||
return 1, 1.24
|
||||
}
|
||||
|
||||
// insertUsageLog inserts a usage_log entry and decrements the spool's remaining grams.
|
||||
func insertUsageLog(ctx context.Context, pool *pgxpool.Pool, jobID, spoolID int, deltaMM, densityGCm3 float64) error {
|
||||
const crossSectionCm2 = 0.02405 // π * (0.0875cm)² for 1.75mm filament
|
||||
gramsUsed := crossSectionCm2 * (deltaMM / 10.0) * densityGCm3
|
||||
|
||||
if gramsUsed <= 0 || deltaMM <= 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
if _, err := pool.Exec(ctx, `
|
||||
INSERT INTO usage_logs (print_job_id, filament_spool_id, mm_extruded, grams_used, logged_at)
|
||||
VALUES ($1, $2, $3, $4, NOW())
|
||||
`, jobID, spoolID, deltaMM, gramsUsed); err != nil {
|
||||
return fmt.Errorf("usage_log insert failed: %w", err)
|
||||
}
|
||||
|
||||
_, _ = pool.Exec(ctx, `
|
||||
UPDATE filament_spools
|
||||
SET remaining_grams = GREATEST(remaining_grams - $2::int, 0),
|
||||
updated_at = NOW()
|
||||
WHERE id = $1
|
||||
`, spoolID, int(gramsUsed))
|
||||
|
||||
slog.Debug("moonraker poller: logged usage",
|
||||
"job_id", jobID, "spool_id", spoolID,
|
||||
"mm_extruded", deltaMM, "grams_used", gramsUsed,
|
||||
)
|
||||
return nil
|
||||
}
|
||||
170
backend/internal/workers/mqtt_subscriber.go
Normal file
170
backend/internal/workers/mqtt_subscriber.go
Normal file
@@ -0,0 +1,170 @@
|
||||
package workers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/CubeCraft-Creations/Extrudex/backend/internal/clients"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// bambuJobState tracks the active print job detected via MQTT.
|
||||
type bambuJobState struct {
|
||||
gcodeFile string
|
||||
gcodeState string
|
||||
percent int
|
||||
}
|
||||
|
||||
// MQTTSubscriber listens to Bambu Lab MQTT telemetry topics and logs
|
||||
// filament usage events to PostgreSQL.
|
||||
type MQTTSubscriber struct {
|
||||
Client *clients.MQTTClient
|
||||
pool *pgxpool.Pool
|
||||
|
||||
printerID int
|
||||
printerName string
|
||||
|
||||
mu sync.Mutex
|
||||
state bambuJobState
|
||||
}
|
||||
|
||||
// MQTTSubscriberConfig holds configuration for the MQTT subscriber worker.
|
||||
type MQTTSubscriberConfig struct {
|
||||
Pool *pgxpool.Pool
|
||||
PrinterID int
|
||||
PrinterName string
|
||||
}
|
||||
|
||||
// NewMQTTSubscriber creates a new MQTTSubscriber worker. Set Client after
|
||||
// construction to wire the handler.
|
||||
func NewMQTTSubscriber(cfg MQTTSubscriberConfig) *MQTTSubscriber {
|
||||
return &MQTTSubscriber{
|
||||
pool: cfg.Pool,
|
||||
printerID: cfg.PrinterID,
|
||||
printerName: cfg.PrinterName,
|
||||
}
|
||||
}
|
||||
|
||||
// Run connects to MQTT and blocks until ctx is cancelled.
|
||||
func (w *MQTTSubscriber) Run(ctx context.Context) error {
|
||||
slog.Info("mqtt subscriber: starting",
|
||||
"printer_id", w.printerID,
|
||||
"printer_name", w.printerName,
|
||||
)
|
||||
|
||||
if w.Client == nil {
|
||||
return fmt.Errorf("mqtt subscriber: Client is nil")
|
||||
}
|
||||
|
||||
if err := w.Client.Connect(); err != nil {
|
||||
return fmt.Errorf("mqtt subscriber: connect failed: %w", err)
|
||||
}
|
||||
defer w.Client.Disconnect()
|
||||
|
||||
slog.Info("mqtt subscriber: connected", "printer_id", w.printerID)
|
||||
|
||||
<-ctx.Done()
|
||||
slog.Info("mqtt subscriber: shutting down", "printer_id", w.printerID)
|
||||
return nil
|
||||
}
|
||||
|
||||
// HandleBambuReport is the MQTT callback for Bambu telemetry messages.
|
||||
func (w *MQTTSubscriber) HandleBambuReport(report clients.BambuPrintReport) error {
|
||||
w.mu.Lock()
|
||||
prev := w.state
|
||||
current := bambuJobState{
|
||||
gcodeFile: report.Print.GcodeFile,
|
||||
gcodeState: report.Print.GcodeState,
|
||||
percent: report.Print.McPercent,
|
||||
}
|
||||
w.state = current
|
||||
w.mu.Unlock()
|
||||
|
||||
if prev.gcodeState == current.gcodeState && prev.gcodeFile == current.gcodeFile {
|
||||
return nil
|
||||
}
|
||||
|
||||
slog.Info("mqtt subscriber: state change",
|
||||
"printer_id", w.printerID,
|
||||
"file", current.gcodeFile,
|
||||
"state", current.gcodeState,
|
||||
"percent", current.percent,
|
||||
)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||||
defer cancel()
|
||||
|
||||
switch current.gcodeState {
|
||||
case "RUNNING":
|
||||
return w.handleState(ctx, current, "printing")
|
||||
case "FINISH":
|
||||
return w.handleState(ctx, current, "completed")
|
||||
case "FAILED":
|
||||
return w.handleState(ctx, current, "failed")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (w *MQTTSubscriber) handleState(ctx context.Context, s bambuJobState, status string) error {
|
||||
jobID, err := w.ensurePrintJob(ctx, s.gcodeFile, status)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if status == "completed" || status == "failed" {
|
||||
_, _ = w.pool.Exec(ctx, `
|
||||
UPDATE print_jobs SET
|
||||
job_status_id = (SELECT id FROM job_statuses WHERE name = $2),
|
||||
completed_at = CASE WHEN $2 = 'completed' THEN NOW() ELSE completed_at END,
|
||||
updated_at = NOW()
|
||||
WHERE id = $1
|
||||
`, jobID, status)
|
||||
} else {
|
||||
_, _ = w.pool.Exec(ctx, `
|
||||
UPDATE print_jobs SET
|
||||
job_status_id = (SELECT id FROM job_statuses WHERE name = $2),
|
||||
started_at = COALESCE(started_at, NOW()),
|
||||
updated_at = NOW()
|
||||
WHERE id = $1
|
||||
`, jobID, status)
|
||||
}
|
||||
|
||||
slog.Info("mqtt subscriber: job updated",
|
||||
"printer_id", w.printerID, "job_id", jobID, "status", status)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (w *MQTTSubscriber) ensurePrintJob(ctx context.Context, filename, status string) (int, error) {
|
||||
var jobID int
|
||||
err := w.pool.QueryRow(ctx, `
|
||||
SELECT id FROM print_jobs
|
||||
WHERE printer_id = $1 AND file_name = $2 AND deleted_at IS NULL
|
||||
ORDER BY created_at DESC LIMIT 1
|
||||
`, w.printerID, filename).Scan(&jobID)
|
||||
|
||||
if err == nil {
|
||||
return jobID, nil
|
||||
}
|
||||
|
||||
var statusID int
|
||||
err = w.pool.QueryRow(ctx, `SELECT id FROM job_statuses WHERE name = $1`, status).Scan(&statusID)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("mqtt subscriber: unknown status '%s': %w", status, err)
|
||||
}
|
||||
|
||||
err = w.pool.QueryRow(ctx, `
|
||||
INSERT INTO print_jobs (printer_id, job_name, file_name, job_status_id, started_at)
|
||||
VALUES ($1, $2, $2, $3, NOW())
|
||||
RETURNING id
|
||||
`, w.printerID, filename, statusID).Scan(&jobID)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("mqtt subscriber: create print_job failed: %w", err)
|
||||
}
|
||||
|
||||
slog.Info("mqtt subscriber: created print job",
|
||||
"job_id", jobID, "file", filename, "status", status)
|
||||
return jobID, nil
|
||||
}
|
||||
Reference in New Issue
Block a user