Elect a single MQTT consumer among replicas

Deployed as a StatefulSet with three replicas, and MQTT has no queue
groups — every subscriber receives every message. All three pods would
commit the same bill and publish three acks. Nothing double-counts,
because the ingest deduplicates on the till's UUID and holds an advisory
lock, but it is three times the database work and three times the
traffic for one sale.

Ordinal 0 consumes; the others stay idle. A StatefulSet already
guarantees stable unique ordinals, so this is a deterministic election
with no lock, no lease and no new dependency. If that pod dies the set
recreates it and tills hold their bills meanwhile, which is what they
are built to do.

POS_MQTT_CONSUMER=always/never overrides it for deployments that are not
a StatefulSet. Anything without an ordinal name — a Deployment pod, a
bare container, local development — consumes, because a lone instance
that silently refused to would be a far more confusing failure than one
that did.

The client id now defaults to the pod name rather than a constant. Two
connections sharing an id evict each other in a reconnect loop that
looks exactly like a flapping network, and takes a while to recognise as
anything else.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

messaging/posmqtt_test.go

@@ -380,3 +380,44 @@ func TestTopicIdentityRejectsShortTopics(t *testing.T) {
 		t.Errorf("topicIdentity = %q/%q, want 12/T4A9", store, terminal)
 	}
 }
+
+// MQTT has no queue groups: every subscriber gets every message. Three replicas
+// all consuming would commit the same bill three times and publish three acks —
+// harmless, because the ingest is idempotent, but three times the work.
+func TestOnlyTheFirstStatefulSetReplicaConsumes(t *testing.T) {
+	cases := []struct {
+		name     string
+		hostname string
+		override string
+		want     bool
+	}{
+		{"statefulset ordinal 0", "fiesta-0", "", true},
+		{"statefulset ordinal 1", "fiesta-1", "", false},
+		{"statefulset ordinal 2", "fiesta-2", "", false},
+		{"double-digit ordinal", "fiesta-10", "", false},
+
+		// A Deployment pod has a random suffix, not an ordinal. Refusing to
+		// consume there would be a far more confusing failure than consuming.
+		{"deployment pod", "fiesta-7d4f9c8b6d-x2k9p", "", true},
+		{"bare container", "a1b2c3d4e5f6", "", true},
+		{"no hostname", "", "", true},
+
+		// The override settles it outright either way.
+		{"forced on", "fiesta-2", "always", true},
+		{"forced off", "fiesta-0", "never", false},
+		{"forced on via true", "fiesta-5", "true", true},
+		{"forced off via false", "fiesta-0", "false", false},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			t.Setenv("HOSTNAME", c.hostname)
+			t.Setenv("POS_MQTT_CONSUMER", c.override)
+
+			if got := posConsumerElected(); got != c.want {
+				t.Errorf("posConsumerElected() = %v, want %v (hostname %q, override %q)",
+					got, c.want, c.hostname, c.override)
+			}
+		})
+	}
+}