nearle_daily/backend_fiesta
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Answer the catalogue as a delta when the terminal sends a revision

Browse Source 
Every pull was a full snapshot, so a shop with a thousand products
re-sent all of them to correct one price. The response now carries a
revision the terminal stores and hands back, and a pull that supplies
one gets only what moved: the product row, its row at that outlet, or
its stock ledger. Stock is included because a shop's count 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 dangerous part is the flag, not the filter. A response marked
is_delta:false tells the terminal to withdraw every product it does not
mention — so a filtered result carrying that label empties the shelf.
Both are now derived from one value, and there is no path through the
function that filters without also setting the flag.

Everything ambiguous resolves toward the snapshot. A revision that is
malformed, empty, or issued to another outlet yields a zero cutoff and a
complete response; the opposite would leave a terminal permanently
missing changes with nothing to show for it. The revision advances only
on the final page, so a terminal that abandons a paginated pull cannot
end up holding one that claims it saw pages it never received. And the
stamp is taken a second in the past, because a product written during
the same second the query ran would otherwise fall on the wrong side of
the next cutoff and be skipped for good.

A delta still cannot withdraw a deleted product — removing a row from
productlocations leaves no tombstone — so a periodic pull without a
revision is what collects those.

Verified against the live outlet: a full pull of 12, a delta returning
only the one product whose price had changed, and pagination that stays
exact now that productid <= 0 is excluded in SQL rather than after the
LIMIT.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Suriya
2026-08-03 19:26:01 +05:30
parent 8709556704
commit fc81df14e4
4 changed files with 250 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
POS_TERMINAL_INGEST.md
View File

@@ -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

150
repositories/posRepository.go
View File

@@ -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,
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
}

67
repositories/posRepository_test.go
View File

@@ -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)
}
}
}

20
scratch/dbinspect/main.go
View File

@@ -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")