diff --git a/POS_TERMINAL_INGEST.md b/POS_TERMINAL_INGEST.md index 81f93dd..cf4e7a4 100644 --- a/POS_TERMINAL_INGEST.md +++ b/POS_TERMINAL_INGEST.md @@ -192,13 +192,45 @@ Worth knowing before the first bill lands. Registrations are **insert-if-absent** — never an update, so a profile corrected at head office is not reverted by a terminal replaying an old capture. -- **Catalogue** answers `is_delta: false` and is therefore a full snapshot. The - terminal withdraws every product a snapshot omits, so this must stay true - while the query returns everything stocked at the outlet. +- **Catalogue** answers a snapshot or a change set, decided by the `since` + revision — see below. - **Barcodes** come from `products.productsku` — there is no barcode column. Scanning at the till matches on it, so SKUs must be the scannable code for barcode scanning to work. +## Catalogue: snapshots and deltas + +`GET /catalogue?store_id=1135` with no `since` returns a **full snapshot**. The +response carries a `revision`; the terminal stores it and sends it back next +time as `since=`, and then gets only what changed. + +A product is included in a change set when any of three things moved: the +product row (name, tax, brand), its row at this outlet (price, availability), or +its stock ledger. Stock counts because a shop's figure drifts from a till's on +every sale rung at another counter, and a delta that ignored it would let that +drift persist until someone forced a full pull. + +**The one rule that matters.** A response marked `is_delta: false` is treated as +a snapshot, and the terminal **withdraws every product it does not mention**. A +filtered result labelled `false` therefore empties the shop's shelf. The filter +and the flag are computed from a single value in `Catalogue()` — there is no +path that filters without also setting the flag, and that is deliberate. + +**A revision that cannot be read falls back to a full snapshot.** Malformed, +empty, or issued to a different outlet — all yield a zero cutoff and a complete +response. The other direction would leave a terminal permanently missing every +change it had not already seen, with nothing to indicate it. + +**The revision only advances on the final page.** A terminal that abandons a +paginated pull half way gets back the revision it already had — or an empty one, +meaning the next pull is a snapshot. Both are recoverable; a prematurely +advanced revision is not. + +**A delta cannot withdraw a deleted product.** A row removed from +`productlocations` leaves no tombstone, so nothing tells the change set to +retire it. Only a snapshot collects those, which is why a terminal should pull +without a revision periodically — the morning import is the natural moment. + ## Terminal health Every till publishes a heartbeat to `nearle/pos/{loc}/{terminal}/health` every @@ -255,8 +287,6 @@ state. ## Not built -- **Catalogue deltas.** Every pull is a full snapshot. Fine for a few hundred - products, worth revisiting at a few thousand. - **Loyalty coming back down.** The uplink deliberately carries no points or spend — those belong to the bill stream, which is idempotent and sees every counter. Nothing yet computes them centrally and sends them to the tills, so diff --git a/repositories/posRepository.go b/repositories/posRepository.go index 5bba429..a23f992 100644 --- a/repositories/posRepository.go +++ b/repositories/posRepository.go @@ -494,12 +494,67 @@ func (r *posRepository) upsertPosCustomer( return nil } -// Catalogue answers a terminal's morning pull. +// posRevisionLayout is the timestamp inside a catalogue revision. // -// Always a full snapshot today, and it says so. The terminal withdraws every -// product a snapshot omits, so answering a change set with is_delta false would -// empty the shelf — declaring false here is only safe because this really does -// return everything stocked at the outlet. +// The revision is the terminal's memory of when it last pulled: it stores what +// we send and hands it back on the next request, and the time encoded in it is +// the cutoff for what has changed since. Colons are avoided so the whole string +// stays safe in a URL query without escaping. +const posRevisionLayout = "20060102T150405Z" + +// posRevisionFor mints the revision a terminal will send back to us. +func posRevisionFor(locationID int, at time.Time) string { + return fmt.Sprintf("loc%d-%s", locationID, at.UTC().Format(posRevisionLayout)) +} + +// posRevisionCutoff reads the timestamp back out of a revision. +// +// Returns the zero time when the revision is missing, malformed, or belongs to +// a different outlet — and a zero cutoff means "send everything". Falling back +// to a full snapshot is the only safe direction: answering an unreadable +// revision with a *delta* would leave the terminal quietly missing every change +// it had not already seen, with nothing to indicate it. +func posRevisionCutoff(locationID int, revision string) time.Time { + revision = strings.TrimSpace(revision) + prefix := fmt.Sprintf("loc%d-", locationID) + if !strings.HasPrefix(revision, prefix) { + return time.Time{} + } + + at, err := time.Parse(posRevisionLayout, strings.TrimPrefix(revision, prefix)) + if err != nil { + return time.Time{} + } + return at +} + +// Catalogue answers a terminal's pull, as a snapshot or as a change set. +// +// ### The rule this function exists to keep +// +// A response with `is_delta: false` is treated as a full snapshot, and the +// terminal **withdraws every product the response does not mention**. So a +// filtered result labelled `false` empties the shop's shelf. +// +// The two are therefore decided together, from one value: a zero cutoff means +// no filter and `is_delta: false`; a non-zero cutoff means filtered and +// `is_delta: true`. There is no path through this function that filters without +// also setting the flag. +// +// ### What counts as a change +// +// A product is included when any of three things moved since the cutoff: the +// product row itself (name, tax, brand), its row at this location (price, +// availability), or its stock ledger. Stock is included because a shop's count +// drifts from the till's on every sale rung elsewhere, and a delta that omitted +// it would let that drift persist until someone forced a full pull. +// +// ### What a delta cannot do +// +// A product *deleted* from productlocations leaves no tombstone, so a change set +// cannot know to withdraw it. Only a full snapshot collects those. A terminal +// should pull without a revision periodically — the morning import is the +// natural moment — and this is why. func (r *posRepository) Catalogue(storeID, since string, page, pageSize int) (*models.PosCatalogueResponse, error) { ctx, err := r.resolvePosStore(storeID) if err != nil { @@ -513,6 +568,11 @@ func (r *posRepository) Catalogue(storeID, since string, page, pageSize int) (*m page = 0 } + // The single decision. Everything downstream reads this rather than + // re-deriving it, so the filter and the flag cannot disagree. + cutoff := posRevisionCutoff(ctx.Locationid, since) + isDelta := !cutoff.IsZero() + type row struct { Productid int Productname string @@ -527,8 +587,25 @@ func (r *posRepository) Catalogue(storeID, since string, page, pageSize int) (*m Status string } + // A product counts as changed if the product row, its row at this location, + // or its stock ledger moved. Written as one predicate so a delta cannot + // miss a price change simply because the product row was untouched. + changed := "" + params := []interface{}{ctx.Tenantid, ctx.Locationid} + if isDelta { + changed = `AND ( + a.updated >= ? + OR b.updated >= ? + OR EXISTS (SELECT 1 FROM productstocks s2 + WHERE s2.productid = a.productid AND s2.tenantid = a.tenantid + AND s2.locationid = b.locationid + AND (s2.stockdate >= ? OR s2.updated >= ?)) + )` + params = append(params, cutoff, cutoff, cutoff, cutoff) + } + rows := make([]row, 0) - err = r.db.Raw(` + query := fmt.Sprintf(` SELECT a.productid, COALESCE(a.productname, '') AS productname, COALESCE(a.productsku, '') AS productsku, @@ -548,12 +625,15 @@ func (r *posRepository) Catalogue(storeID, since string, page, pageSize int) (*m FROM products a INNER JOIN productlocations b ON a.productid = b.productid AND a.tenantid = b.tenantid LEFT JOIN productcategories c ON a.categoryid = c.categoryid - WHERE a.tenantid = ? AND b.locationid = ? + WHERE a.tenantid = ? AND b.locationid = ? AND a.productid > 0 %s ORDER BY a.productid - LIMIT ? OFFSET ?`, - ctx.Tenantid, ctx.Locationid, pageSize+1, page*pageSize, - ).Scan(&rows).Error - if err != nil { + LIMIT ? OFFSET ?`, changed) + + // One row past the page, purely so has_more can be answered without a + // second count query. + params = append(params, pageSize+1, page*pageSize) + + if err := r.db.Raw(query, params...).Scan(&rows).Error; err != nil { return nil, err } @@ -566,14 +646,12 @@ func (r *posRepository) Catalogue(storeID, since string, page, pageSize int) (*m products := make([]models.PosCatalogueProduct, 0, len(rows)) for _, p := range rows { - // A productid of zero is bad data, not a product — live data has at - // least one, almost certainly an insert that never got a sequence - // value. Sending it would put a row on the till that can never be - // billed, because the ingest refuses any line whose id is not positive. - if p.Productid <= 0 { - continue - } - + // Rows with productid <= 0 are excluded in SQL rather than here. Live + // data has at least one — almost certainly an insert that never got a + // sequence value — and it can never be billed, because the ingest + // refuses any line whose id is not positive. Filtering it in the query + // also keeps pagination exact: skipped after the LIMIT, it would eat a + // slot and hand back a short page. mrp := p.Retailprice if mrp <= p.Price { mrp = 0 @@ -612,15 +690,39 @@ func (r *posRepository) Catalogue(storeID, since string, page, pageSize int) (*m }) } + // The revision only advances on the final page. + // + // A terminal that gives up half way through a paginated pull — a dropped + // connection, a till switched off — must not be left holding a revision + // that claims it has seen pages it never received. Every one of those + // products would then be excluded from the next delta and stay stale + // indefinitely, with nothing anywhere to indicate it. + // + // So mid-pull we echo back whatever the terminal already had: unchanged if + // it sent one, empty if it did not, and empty means the next pull is a full + // snapshot. Both are recoverable; a prematurely advanced revision is not. + // + // The stamp is taken a second in the past. A product written during the + // same second this query ran could otherwise land on the wrong side of the + // next cutoff and be skipped for good — overlapping by a second costs one + // redundant row and cannot lose one. + revision := strings.TrimSpace(since) + if !hasMore { + revision = posRevisionFor(ctx.Locationid, time.Now().Add(-time.Second)) + } + return &models.PosCatalogueResponse{ - // A revision the terminal stores and sends back on its next pull. Tied - // to the outlet and the moment, so a shop that has pulled today can be - // told it is already current. - Revision: fmt.Sprintf("loc%d-%s", ctx.Locationid, time.Now().UTC().Format("20060102T150405")), - Isdelta: false, - Hasmore: hasMore, - Products: products, - Customers: make([]models.PosCatalogueCustomer, 0), + Revision: revision, + // Decided with the filter, never separately. False here would tell the + // terminal to withdraw every product this response omits. + Isdelta: isDelta, + Hasmore: hasMore, + Products: products, + Customers: make([]models.PosCatalogueCustomer, 0), + // A product deleted from productlocations leaves no tombstone, so a + // change set cannot know to withdraw it. Only a full snapshot collects + // those, which is why a terminal should pull without a revision + // periodically. Retiredids: make([]string, 0), }, nil } diff --git a/repositories/posRepository_test.go b/repositories/posRepository_test.go index 425c2c1..f11a5cf 100644 --- a/repositories/posRepository_test.go +++ b/repositories/posRepository_test.go @@ -1,6 +1,9 @@ package repositories -import "testing" +import ( + "testing" + "time" +) // The terminal holds a unique index on barcode, so this rule decides whether a // catalogue import succeeds at all. Measured against live data when it was @@ -95,3 +98,65 @@ func TestLegacyOrderQtyIsUnchanged(t *testing.T) { } } } + +// A catalogue revision is the terminal's memory of when it last pulled. If it +// does not survive a round trip, every pull silently becomes a full snapshot — +// or worse, a filtered result gets labelled as one and the shop's shelf empties. +func TestPosRevisionRoundTrips(t *testing.T) { + at := time.Date(2026, 8, 3, 12, 30, 45, 0, time.UTC) + + revision := posRevisionFor(1135, at) + if revision != "loc1135-20260803T123045Z" { + t.Fatalf("revision = %q, want loc1135-20260803T123045Z", revision) + } + + got := posRevisionCutoff(1135, revision) + if !got.Equal(at) { + t.Errorf("cutoff = %v, want %v", got, at) + } +} + +func TestAnUnusableRevisionFallsBackToAFullSnapshot(t *testing.T) { + // A zero cutoff means "send everything", and the caller turns that into + // is_delta:false. Falling back the other way — answering an unreadable + // revision with a change set — would leave a terminal permanently missing + // every change it had not already seen, with nothing to show for it. + cases := []struct { + name string + location int + revision string + }{ + {"empty", 1135, ""}, + {"whitespace", 1135, " "}, + {"no prefix", 1135, "20260803T123045Z"}, + {"malformed timestamp", 1135, "loc1135-not-a-time"}, + {"truncated timestamp", 1135, "loc1135-20260803"}, + {"another outlet's revision", 1135, "loc1097-20260803T123045Z"}, + {"prefix collision", 113, "loc1135-20260803T123045Z"}, + {"garbage", 1135, "../../etc/passwd"}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := posRevisionCutoff(c.location, c.revision); !got.IsZero() { + t.Errorf("cutoff = %v, want zero (full snapshot) for %q", got, c.revision) + } + }) + } +} + +func TestAnOutletCannotReplayAnotherOutletsRevision(t *testing.T) { + // loc1135 and loc113 share a textual prefix. Matching loosely would let one + // shop's cutoff silently scope another shop's delta. + at := time.Date(2026, 8, 3, 12, 30, 45, 0, time.UTC) + revision := posRevisionFor(1135, at) + + if got := posRevisionCutoff(1135, revision); got.IsZero() { + t.Error("the issuing outlet could not read back its own revision") + } + for _, other := range []int{113, 11350, 1097, 1} { + if got := posRevisionCutoff(other, revision); !got.IsZero() { + t.Errorf("outlet %d accepted outlet 1135's revision (cutoff %v)", other, got) + } + } +} diff --git a/scratch/dbinspect/main.go b/scratch/dbinspect/main.go index 966ea73..5d57985 100644 --- a/scratch/dbinspect/main.go +++ b/scratch/dbinspect/main.go @@ -153,6 +153,26 @@ func main() { case "cleanup": cleanup(db) + case "columns": + for _, t := range []string{"products", "productlocations", "productstocks"} { + fmt.Printf("=== %s ===\n", t) + var cols []struct { + ColumnName string + DataType string + } + db.Raw(`SELECT column_name, data_type FROM information_schema.columns + WHERE table_name = ? ORDER BY ordinal_position`, t).Scan(&cols) + for _, c := range cols { + marker := "" + n := c.ColumnName + if n == "created" || n == "updated" || n == "updated_at" || + n == "stockdate" || n == "modified" { + marker = " <-- timestamp" + } + fmt.Printf(" %-24s %s%s\n", c.ColumnName, c.DataType, marker) + } + } + case "customer": fmt.Println("=== customers matching the uplink probe ===") showCustomer(db, "9840012345")