backend_fiesta/messaging/posmqtt.go

package messaging

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"strconv"
	"strings"
	"time"

	"nearle/models"
	"nearle/services"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

// Plain-MQTT ingest for the Nearle POS terminals.
//
// The sibling of posconsumer.go, and which one you want depends entirely on
// what is listening on the other end:
//
//   - **This file** talks MQTT to a broker like Mosquitto or EMQX — the kind
//     already running at the rider app's `66.116.225.226:1883`.
//   - **posconsumer.go** talks the NATS protocol to a NATS server, which
//     exposes MQTT through a gateway but speaks NATS itself on 4222.
//
// They are not interchangeable: a NATS client cannot connect to Mosquitto, and
// an MQTT client cannot use NATS' native subjects. Both call the same
// PosService, so whichever is running, a bill lands identically.
//
// Enabled with MQTT_URL. Both may run at once, which is what a migration
// between brokers looks like.

const (
	// Namespaced under `nearle/` alongside the rider app's
	// `nearle/riders/{riderId}/...`, so one broker ACL rule covers each system
	// and it is obvious from a topic which one it belongs to.
	//
	// Wildcards for MQTT are `+` per level, where NATS uses `*`.
	topicOrders    = "nearle/pos/+/+/order"
	topicCustomers = "nearle/pos/+/+/customer"
	topicHealth    = "nearle/pos/+/+/health"
)

type PosMqttConsumer struct {
	client mqtt.Client
	svc    services.PosService
}

// StartPosMqttConsumer connects and subscribes.
//
// Returns (nil, nil) when MQTT_URL is unset — a deployment without a broker is
// supported, and the caller carries on with the HTTP endpoints.
func StartPosMqttConsumer(svc services.PosService) (*PosMqttConsumer, error) {
	url := strings.TrimSpace(os.Getenv("MQTT_URL"))
	if url == "" {
		log.Println("pos: MQTT_URL not set, plain-MQTT ingest disabled")
		return nil, nil
	}

	c := &PosMqttConsumer{svc: svc}

	opts := mqtt.NewClientOptions().
		AddBroker(url).
		// Stable, so the broker resumes this session and redelivers anything
		// in flight rather than treating every restart as a new subscriber.
		SetClientID(getEnvDefault("MQTT_CLIENT_ID", "nearle-pos-ingest")).
		SetCleanSession(false).
		SetAutoReconnect(true).
		SetMaxReconnectInterval(30 * time.Second).
		SetKeepAlive(30 * time.Second).
		SetConnectionLostHandler(func(_ mqtt.Client, err error) {
			log.Printf("pos: MQTT connection lost: %v", err)
		})

	if user := os.Getenv("MQTT_USER"); user != "" {
		opts.SetUsername(user).SetPassword(os.Getenv("MQTT_PASSWORD"))
	}

	// Re-subscribed on every (re)connect rather than once at startup: with a
	// broker that did not persist the session, a reconnect would otherwise come
	// back silently subscribed to nothing.
	opts.SetOnConnectHandler(func(client mqtt.Client) {
		log.Printf("pos: connected to MQTT broker %s", url)
		for topic, handler := range map[string]mqtt.MessageHandler{
			topicOrders:    c.handleOrders,
			topicCustomers: c.handleCustomers,
			topicHealth:    c.handleHealth,
		} {
			if token := client.Subscribe(topic, 1, handler); token.Wait() && token.Error() != nil {
				log.Printf("pos: could not subscribe to %s: %v", topic, token.Error())
				continue
			}
			log.Printf("pos: subscribed to %s", topic)
		}
	})

	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		return nil, fmt.Errorf("could not connect to the MQTT broker at %s: %w", url, token.Error())
	}
	c.client = client

	return c, nil
}

func (c *PosMqttConsumer) handleOrders(_ mqtt.Client, msg mqtt.Message) {
	var batch models.PosOrderBatch
	if err := json.Unmarshal(msg.Payload(), &batch); err != nil {
		// Dropped rather than retried: there is no batch id to answer with, and
		// the till will time out and re-send anyway.
		log.Printf("pos: discarding unreadable order batch on %s: %v", msg.Topic(), err)
		return
	}

	store, terminal := topicIdentity(msg.Topic())
	if batch.Storeid == "" {
		batch.Storeid = store
	}
	if batch.Terminalid == "" {
		batch.Terminalid = terminal
	}

	ack, err := c.svc.IngestOrders(batch)
	if err != nil {
		// Nothing committed, so nothing is acknowledged. The terminal keeps
		// every bill and retries — which is the entire point of the design.
		log.Printf("pos: order batch %s from %s/%s failed, not acking: %v",
			batch.Batchid, store, terminal, err)
		return
	}

	c.publishAck(store, terminal, ack)
	log.Printf("pos: order batch %s from %s/%s — %d accepted, %d rejected",
		batch.Batchid, store, terminal, len(ack.Accepted), len(ack.Rejected))
}

func (c *PosMqttConsumer) handleCustomers(_ mqtt.Client, msg mqtt.Message) {
	var batch models.PosCustomerBatch
	if err := json.Unmarshal(msg.Payload(), &batch); err != nil {
		log.Printf("pos: discarding unreadable customer batch on %s: %v", msg.Topic(), err)
		return
	}

	store, terminal := topicIdentity(msg.Topic())
	if batch.Storeid == "" {
		batch.Storeid = store
	}
	if batch.Terminalid == "" {
		batch.Terminalid = terminal
	}

	ack, err := c.svc.IngestCustomers(batch)
	if err != nil {
		log.Printf("pos: customer batch %s from %s/%s failed, not acking: %v",
			batch.Batchid, store, terminal, err)
		return
	}

	c.publishAck(store, terminal, ack)
}

// handleHealth records one heartbeat.
//
// Never acknowledged. Presence is fire-and-forget: a till whose heartbeat
// failed must carry on selling, and a blank square on a dashboard is a far
// better outcome than a terminal that stopped because Redis was busy.
func (c *PosMqttConsumer) handleHealth(_ mqtt.Client, msg mqtt.Message) {
	var health models.PosHealth
	if err := json.Unmarshal(msg.Payload(), &health); err != nil {
		log.Printf("pos: discarding unreadable heartbeat on %s: %v", msg.Topic(), err)
		return
	}

	// From the topic, not the body — the same rule bills follow.
	store, terminal := topicIdentity(msg.Topic())
	if health.Locationid == "" {
		health.Locationid = store
	}
	if health.Terminalid == "" {
		health.Terminalid = terminal
	}

	// The broker's Last Will arrives here too, as a bare {"status":"offline"}
	// with no other fields, which is exactly what should be recorded when a
	// till loses power mid-shift.
	if health.Status == "" {
		health.Status = "online"
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	if err := c.svc.RecordHealth(ctx, health); err != nil {
		log.Printf("pos: could not record heartbeat from %s/%s: %v", store, terminal, err)
	}
}

// publishAck answers the till that sent the batch, and only that till.
func (c *PosMqttConsumer) publishAck(store, terminal string, ack *models.PosAck) {
	if store == "" || terminal == "" {
		log.Printf("pos: cannot ack batch %s — the topic named no terminal", ack.Batchid)
		return
	}

	payload, err := json.Marshal(ack)
	if err != nil {
		log.Printf("pos: could not encode ack for batch %s: %v", ack.Batchid, err)
		return
	}

	topic := fmt.Sprintf("nearle/pos/%s/%s/ack", store, terminal)

	// QoS 1: losing an ack means the till re-sends bills that are already
	// banked. Harmless, because the ingest deduplicates — but wasted traffic on
	// a shop line that may not have much to spare.
	token := c.client.Publish(topic, 1, false, payload)
	if !token.WaitTimeout(10*time.Second) || token.Error() != nil {
		log.Printf("pos: could not publish ack to %s: %v", topic, token.Error())
	}
}

// topicIdentity reads the store and terminal out of
// `nearle/pos/<store>/<terminal>/<kind>`.
//
// Taken from the topic rather than the body on purpose: a till that could name
// a store in its payload could post sales into another shop's books.
func topicIdentity(topic string) (store, terminal string) {
	parts := strings.Split(topic, "/")
	if len(parts) < 5 {
		return "", ""
	}
	return parts[2], parts[3]
}

// PublishCatalogueChanged tells every till in a store to pull now.
//
// Retained, so a terminal that was switched off during the change still hears
// about it when it comes back.
func (c *PosMqttConsumer) PublishCatalogueChanged(storeID, revision string) error {
	payload, err := json.Marshal(map[string]string{"revision": revision})
	if err != nil {
		return err
	}
	token := c.client.Publish(fmt.Sprintf("nearle/pos/%s/catalogue", storeID), 1, true, payload)
	token.Wait()
	return token.Error()
}

// Close disconnects, allowing a moment for in-flight acks to leave.
func (c *PosMqttConsumer) Close() {
	if c == nil || c.client == nil {
		return
	}
	quiesce, err := strconv.Atoi(getEnvDefault("MQTT_QUIESCE_MS", "2000"))
	if err != nil || quiesce < 0 {
		quiesce = 2000
	}
	c.client.Disconnect(uint(quiesce))
}

func getEnvDefault(key, fallback string) string {
	if v := strings.TrimSpace(os.Getenv(key)); v != "" {
		return v
	}
	return fallback
}