86 lines
3.1 KiB
Go
86 lines
3.1 KiB
Go
package repositories
|
|
|
|
import (
|
|
"log"
|
|
"strconv"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// Telling tills that a shop's shelf has moved.
|
|
//
|
|
// The broker topic `nearle/pos/{loc}/catalogue` and its publisher have existed
|
|
// since the POS integration landed, retained flag and all — and **nothing ever
|
|
// called it**. Every terminal's stock figure was therefore whatever it last
|
|
// pulled, with no signal that an app order had just sold the last of something.
|
|
//
|
|
// It could not be called from here, which is why it never was: `repositories`
|
|
// does not import `messaging`, and must not — wiring runs the other way, from
|
|
// main.go, and reversing it would be an import cycle. So the dependency is
|
|
// inverted through this interface. `messaging.PosMqttConsumer` already
|
|
// satisfies it; main.go registers it once at startup.
|
|
//
|
|
// A package-level registration rather than a constructor parameter because the
|
|
// two write paths that need it — POS bill ingest and createOrderTx — are
|
|
// reached through several layers that would each have to grow a field for a
|
|
// notification neither of them cares about the result of.
|
|
|
|
// CatalogueNotifier tells every till at a store to pull the catalogue again.
|
|
type CatalogueNotifier interface {
|
|
PublishCatalogueChanged(storeID, revision string) error
|
|
}
|
|
|
|
var (
|
|
catalogueNotifierMu sync.RWMutex
|
|
catalogueNotifier CatalogueNotifier
|
|
)
|
|
|
|
// SetCatalogueNotifier registers the publisher. Safe to leave unset — the API
|
|
// must run with the broker down, and a missed notification only means a till is
|
|
// stale until its next periodic pull.
|
|
func SetCatalogueNotifier(n CatalogueNotifier) {
|
|
catalogueNotifierMu.Lock()
|
|
defer catalogueNotifierMu.Unlock()
|
|
catalogueNotifier = n
|
|
}
|
|
|
|
// notifyCatalogueChanged asks the tills at one outlet to refresh.
|
|
//
|
|
// Three rules, all of them learned the hard way elsewhere in this codebase:
|
|
//
|
|
// 1. **Call it after the transaction commits, never inside.** A rolled-back
|
|
// sale must not announce a change that did not happen.
|
|
// 2. **It cannot fail a sale.** A publish error is logged and swallowed —
|
|
// exactly as syncProductLocationStatus already does for the availability
|
|
// flag. Stock is committed; a terminal being told about it late is not
|
|
// worth losing a bill over.
|
|
// 3. **Only the revision travels, never quantities.** The till pulls. Pushed
|
|
// numbers race with concurrent sales, and the catalogue endpoint already
|
|
// answers a delta from a revision.
|
|
//
|
|
// Fire-and-forget on its own goroutine: the publisher waits for the broker's
|
|
// acknowledgement, and a bill's response should not sit behind that.
|
|
func notifyCatalogueChanged(locationID int) {
|
|
if locationID <= 0 {
|
|
return
|
|
}
|
|
|
|
catalogueNotifierMu.RLock()
|
|
n := catalogueNotifier
|
|
catalogueNotifierMu.RUnlock()
|
|
if n == nil {
|
|
return
|
|
}
|
|
|
|
// Stamped a second in the past for the same reason Catalogue() does it: a
|
|
// revision must not claim to include a write that is still landing.
|
|
revision := posRevisionFor(locationID, time.Now().Add(-time.Second))
|
|
storeID := strconv.Itoa(locationID)
|
|
|
|
go func() {
|
|
if err := n.PublishCatalogueChanged(storeID, revision); err != nil {
|
|
log.Printf("catalogue notify: store %s: %v", storeID, err)
|
|
}
|
|
}()
|
|
}
|