backend_fiesta/scratch/mqttpub/main.go

// Publishes a bill over the real broker and waits for the ack, standing in for
// a till until one is available to test with.
//
//	go run ./scratch/mqttpub
package main

import (
	"encoding/json"
	"fmt"
	"log"
	"os"
	"time"

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

const (
	locationID = "1135"
	terminalID = "T4A9"
	orderID    = "99999999-8888-4777-8666-555555555555" // distinct from the HTTP test
)

func main() {
	_ = godotenv.Load()

	broker := os.Getenv("MQTT_URL")
	if broker == "" {
		broker = "tcp://66.116.225.226:1883"
	}

	opts := mqtt.NewClientOptions().
		AddBroker(broker).
		SetClientID("pos-e2e-probe").
		SetUsername(os.Getenv("MQTT_USER")).
		SetPassword(os.Getenv("MQTT_PASSWORD")).
		SetCleanSession(true)

	client := mqtt.NewClient(opts)
	if t := client.Connect(); t.Wait() && t.Error() != nil {
		log.Fatal("connect:", t.Error())
	}
	defer client.Disconnect(500)
	fmt.Println("connected to", broker)

	acks := make(chan []byte, 1)
	ackTopic := fmt.Sprintf("nearle/pos/%s/%s/ack", locationID, terminalID)
	if t := client.Subscribe(ackTopic, 1, func(_ mqtt.Client, m mqtt.Message) {
		acks <- m.Payload()
	}); t.Wait() && t.Error() != nil {
		log.Fatal("subscribe:", t.Error())
	}
	fmt.Println("listening on", ackTopic)

	batch := map[string]any{
		"schema":      1,
		"batch_id":    "batch-mqtt-0001",
		"store_id":    locationID,
		"terminal_id": terminalID,
		"sent_at":     time.Now().UTC().Format(time.RFC3339),
		"orders": []map[string]any{{
			"id":              orderID,
			"invoice_number":  "INV-2608-T4A9-00002",
			"created_at":      time.Now().UTC().Format(time.RFC3339),
			"terminal_id":     terminalID,
			"cashier":         "Divya",
			"customer":        map[string]any{"id": "c-2", "mobile": "9840099999", "name": "Ravi"},
			"subtotal":        60.0,
			"discount":        0.0,
			"promos":          []any{},
			"tax":             4.44,
			"tax_breakdown":   map[string]float64{"0.08": 4.44},
			"round_off":       0.0,
			"total":           60.0,
			"points_earned":   0,
			"points_redeemed": 0,
			"payments":        []map[string]any{{"method": "upi", "amount": 60.0, "reference": "TESTUPI"}},
			"items": []map[string]any{{
				"product_id": "6988", "barcode": "6988", "name": "Mysore Banana",
				"quantity": 1, "unit_price": 60.0, "discount": 0.0,
				"gst_rate": 0.08, "tax": 4.44, "line_total": 60.0,
			}},
		}},
	}

	body, _ := json.Marshal(batch)
	orderTopic := fmt.Sprintf("nearle/pos/%s/%s/order", locationID, terminalID)
	if t := client.Publish(orderTopic, 1, false, body); t.Wait() && t.Error() != nil {
		log.Fatal("publish:", t.Error())
	}
	fmt.Println("published a bill to", orderTopic)

	// The same 20 seconds a real till waits before giving up and re-sending.
	select {
	case payload := <-acks:
		fmt.Println("\nACK RECEIVED:")
		var pretty map[string]any
		_ = json.Unmarshal(payload, &pretty)
		out, _ := json.MarshalIndent(pretty, "  ", "  ")
		fmt.Println("  " + string(out))
	case <-time.After(20 * time.Second):
		fmt.Println("\nNO ACK within 20s — a real till would keep the bill and send it again")
		os.Exit(1)
	}

	// A heartbeat too, so the Redis presence path is exercised.
	health, _ := json.Marshal(map[string]any{
		"schema": 1, "status": "online",
		"terminal_id": terminalID, "location_id": locationID,
		"store_name": "Ragul stores Selvapuram", "app_version": "1.1.0",
		"pending_bills": 0, "pending_registrations": 0,
		"today_bills": 2, "today_amount": 170.0,
		"printer_reachable": false,
		"reported_at":       time.Now().UTC().Format(time.RFC3339),
	})
	healthTopic := fmt.Sprintf("nearle/pos/%s/%s/health", locationID, terminalID)
	if t := client.Publish(healthTopic, 1, false, health); t.Wait() && t.Error() != nil {
		log.Fatal("publish health:", t.Error())
	}
	fmt.Println("\npublished a heartbeat to", healthTopic)

	// The registration uplink. Tested over HTTP early on; this is the same
	// service reached over the broker, which is the path a real till uses.
	custBatch, _ := json.Marshal(map[string]any{
		"schema": 1, "batch_id": "batch-cust-mqtt-0001",
		"store_id": locationID, "terminal_id": terminalID,
		"customers": []map[string]any{{
			"id":                     "3d7a0000-0000-4000-8000-000000000001",
			"mobile":                 "9840077777",
			"name":                   "MQTT Probe Shopper",
			"registered_at":          time.Now().UTC().Format(time.RFC3339),
			"registered_by_terminal": terminalID,
		}},
	})
	custTopic := fmt.Sprintf("nearle/pos/%s/%s/customer", locationID, terminalID)
	if t := client.Publish(custTopic, 1, false, custBatch); t.Wait() && t.Error() != nil {
		log.Fatal("publish customer:", t.Error())
	}
	fmt.Println("published a registration to", custTopic)

	select {
	case payload := <-acks:
		fmt.Println("  registration ACK:", string(payload))
	case <-time.After(20 * time.Second):
		fmt.Println("  NO ACK for the registration")
		os.Exit(1)
	}
}