package dwingest import ( "context" "encoding/json" "fmt" "log" "sync" "time" "git.joco.dk/sng/fermentord/internal/configuration" "git.joco.dk/sng/fermentord/internal/controllers" "git.joco.dk/sng/fermentord/pkg/temperature" "git.joco.dk/sng/fermentord/pkg/tilt" "github.com/getsentry/sentry-go" "github.com/gofrs/uuid" "github.com/nats-io/nats.go" ) type DWIngest struct { chTemperatureReading chan temperature.TemperatureReading chState chan controllers.ChamberState js nats.JetStream hub *sentry.Hub serial int64 } func NewDWIngest(js nats.JetStream) *DWIngest { return &DWIngest{ chTemperatureReading: make(chan temperature.TemperatureReading, 3600), chState: make(chan controllers.ChamberState, 100), hub: sentry.CurrentHub().Clone(), js: js, } } func (p *DWIngest) AddReading(reading temperature.TemperatureReading) { select { case p.chTemperatureReading <- reading: break default: err := fmt.Errorf("channel overflow on dwingest temperature channel") p.hub.CaptureException(err) log.Print(err) } } func (p *DWIngest) AddState(state controllers.ChamberState) { select { case p.chState <- state: break default: err := fmt.Errorf("channel overflow on dwingest state channel") p.hub.CaptureException(err) log.Print(err) } } func (p *DWIngest) Run(ctx context.Context, wg *sync.WaitGroup, config configuration.Configuration) { defer wg.Done() defer p.hub.Flush(10 * time.Second) for { select { case reading := <-p.chTemperatureReading: p.Publish(config.NATS.Subject.Temp, reading) case state := <-p.chState: p.Publish(config.NATS.Subject.State, State{ Time: time.Now().UTC(), State: controllers.ChamberStateMap[state], }) case t := <-tilt.C: p.Publish(config.NATS.Subject.Tilt, Tilt{ Time: time.Now().UTC(), Color: string(t.Color()), Gravity: t.Gravity(), Temperature: t.Celsius(), }) case <-ctx.Done(): return } } } func (p *DWIngest) Publish(subject string, reading any) error { b, err := json.Marshal(reading) if err != nil { p.hub.CaptureException(err) log.Printf("Error marshaling JSON: %v", err) return err } msgID, err := uuid.NewV7(uuid.MillisecondPrecision) if err != nil { p.hub.CaptureException(err) log.Print(err) return err } msg := nats.NewMsg(subject) msg.Header.Add("Nats-Msg-Id", msgID.String()) msg.Data = b _, err = p.js.PublishMsg(msg, nats.AckWait(5*time.Second)) if err != nil { p.hub.CaptureException(err) log.Print(err) return err } return nil }