fermentord/internal/dwingest/nats.go

package dwingest

import (
	"context"
	"encoding/json"
	"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/nats-io/nats.go"
)

type DWIngest struct {
	chTemperatureReading chan temperature.TemperatureReading
	chState              chan controllers.ChamberState
}

func NewDWIngest() *DWIngest {
	return &DWIngest{
		chTemperatureReading: make(chan temperature.TemperatureReading, 3600),
		chState:              make(chan controllers.ChamberState, 100),
	}
}

func (p *DWIngest) AddReading(reading temperature.TemperatureReading) {
	p.chTemperatureReading <- reading
}

func (p *DWIngest) AddState(state controllers.ChamberState) {
	p.chState <- state
}

func (p *DWIngest) Run(ctx context.Context, wg *sync.WaitGroup, js nats.JetStream, config *configuration.Configuration) {
	hub := sentry.CurrentHub().Clone()
	defer hub.Flush(10 * time.Second)
	defer wg.Done()

	for {
		select {
		case reading := <-p.chTemperatureReading:
			publish(config.NATS.Subject.Temp, js, hub, reading)

		case state := <-p.chState:
			reading := State{
				Time:  time.Now().UTC(),
				State: controllers.ChamberStateMap[state],
			}
			publish(config.NATS.Subject.State, js, hub, reading)

		case t := <-tilt.C:
			reading := Tilt{
				Time:        time.Now().UTC(),
				Color:       string(t.Color()),
				Gravity:     t.Gravity(),
				Temperature: t.Celsius(),
			}
			publish(config.NATS.Subject.Tilt, js, hub, reading)

		case <-ctx.Done():
			close(p.chTemperatureReading)
			close(p.chState)

			return
		}
	}
}

func publish(subject string, js nats.JetStream, hub *sentry.Hub, reading any) error {
	b, err := json.Marshal(reading)
	if err != nil {
		hub.CaptureException(err)
		log.Printf("Error marshaling JSON: %v", err)
		return err
	}

	_, err = js.Publish(subject, b)
	if err != nil {
		hub.CaptureException(err)
		log.Printf("Error publishing to NATS: %v", err)
	}

	return err
}