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

Compare commits

base: nearle_daily:27fbbf0422f59c37e3c9da6d7fe2751b919bc0c1
Branches Tags
nearle_daily:main
...
compare: nearle_daily:main
Branches Tags
nearle_daily:main

13 Commits
27fbbf0422 ... main

Author SHA1 Message Date
Suriya
5864204d32 Zero the produce rates, on the owner's instruction 
The eight fresh lines at 1135 held 8, 12 and 18. Fresh unbranded fruit and
chilled fish are nil-rated under Indian GST, so those were overcharging.

Held back on the first pass and reported as REVIEW, because every one is a
reduction of a live tax rate and that is a decision for whoever signs the
returns rather than something a script should quietly do. Put to the owner and
released explicitly.

Two readings are assumed and are worth checking against what the counter
actually sells. Maceral and Tuna are taken as fresh or chilled — frozen,
branded or packaged fish is 5%. Hatsun curd was already 0 and stays there as
plain curd; flavoured yoghurt would be 5%.

The undo SQL for all eight is in the tool's output and restores the previous
rates exactly.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-05 20:51:56 +05:30
Suriya
d0c3cb751e Document the sale-date contract, and rate the packaged goods 
parsePosSaleDate needed no change — RFC3339Nano already accepts the offset the
terminal now sends, and Format("2006-01-02") on a zoned time still yields the
till's own trading day rather than UTC's. But the ordering of those layouts is
load-bearing and nothing said so, and the two bare layouts are a legacy that
should be recognisable as one: they exist for terminals built before the offset,
whose bills record an instant wrong by the offset with nothing in the payload to
recover it from. Two tests pin both halves, including the case that motivated
this — 00:30 IST, where UTC has not yet rolled into the same day.

The GST script writes only the four packaged lines at 1185, which sat at 0 and
were being billed with no tax at all.

It deliberately does not touch the produce at 1135. That was written believing
every rate was 0 — read from a field name that does not exist in the response,
so the check silently returned nothing. The rows in fact hold 8, 12 and 18, and
under Indian GST fresh unbranded fruit and chilled fish are nil-rated, so
several look like overcharging. Every correction there is a reduction of a live
rate, which belongs to whoever signs the returns rather than to a script. They
are reported as REVIEW and left as found.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-05 20:18:17 +05:30
Suriya
11595ad415 Record the probes that touched live data 
seedprices is the only account of what the 15 seeded retail prices were and how
to put the zeros back — products at 1135 and 1185 were all at 0, so nothing was
sellable and the terminal could not be exercised at all. It refuses to overwrite
a price a human already set, so re-running it is safe.

termfixcleanup removes the one bill posted to prove the terminalid fix against
production. Named by its own terminalorderid rather than by date or by "the
newest row": pos_orders holds real takings and is not a table to run an
unbounded DELETE against.

healthproof carries a warning it did not have when it was written. MQTT_USER in
.env is pos_ingest, which the ACL denies publish on the health topic, and
Mosquitto answers an ACL-denied QoS 1 publish with a PUBACK before discarding
it. So the tool connects, reports success, and nothing arrives — indistinguishable
from a dead consumer, which is how it read for an hour.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-05 18:48:24 +05:30
Suriya
ec672a3087 Add an HTTP heartbeat, and file bills under the till that rang them 
Two faults found while checking whether today's live bills had landed. They
had — 17 of them, complete — but both of these were sitting in the same data.

**Health existed only over MQTT.** The consumer subscribes to the health topic
and has done since startup, but a terminal on the HTTP route has no way to
reach it. Today's terminal was on HTTP, so it reported nothing and the board
showed "online 0 of 1" while the till was demonstrably alive and selling.
POST /pos/health now takes the same payload the broker carries, into the same
Redis record, so the board cannot tell the two routes apart and does not need
to. It answers 202 and swallows failures: a till that cannot say how it is must
still sell.

**terminalid was empty on 16 of 17 bills.** The consumer backfills a missing
terminal code from the topic, but onto the batch, while the row was built from
the order — the two never met, and importPosOrder was not handed the batch's
value at all. Over HTTP there was no topic to fall back on either. So the
invoice numbers read INV-2608-T5EDD-000NN while the column they should have
matched was blank, and `byterminal` on the sales summary grouped almost
everything under "". The bill's own terminal now wins with the batch's as the
fallback, trimmed, so whitespace is not mistaken for a code.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-05 18:48:24 +05:30
abhishek
bddd8fa265 product price 2026-08-04 10:58:26 +05:30
Suriya
8aa4d86eb6 Add a POS integration handover 
An API reference and rationale for whoever picks this up next. The
endpoints are the easy half; what is not obvious from reading the code is
why an ack is only published after the commit, why a duplicate counts as
a success, and why is_delta false on a filtered catalogue empties a
shop's shelf. Those are written down here because each one costs a shop
money when someone changes it without knowing.

Covers the GET endpoints in full — request parameters, real response
shapes, and the cases that surprise people: values coming back as
strings from a Redis hash, a quiet till answering 200 rather than 404,
and a bill at another outlet returning 404 even though the reference is
valid.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-03 20:22:11 +05:30
Suriya
b0caacd90a pos 2026-08-03 20:19:04 +05:30
Suriya
b9f389fcdf Process the MQTT ingest on a bounded worker pool 
paho delivers on one goroutine, so bills were committed strictly one
after another. Each is a full Postgres transaction — advisory lock,
dedup, stock row locks, availability check, four inserts, commit — which
is 10-30ms, so the ceiling was roughly 30-100 bills a second and a
shop's backlog draining after an outage took minutes to land.

A fixed pool behind a bounded queue, rather than a goroutine per
message. Unbounded concurrency would open a transaction per message and
exhaust the connection pool under a storm, stalling every one of them at
once — a slow minute turned into a dead one. When the queue fills,
submit blocks: paho stops acknowledging, the broker's in-flight window
fills, it stops sending, and the backpressure reaches the till, which
holds its bills and retries. Slow, but nothing is dropped.

Heartbeats get their own pool. Sharing one would let a backlog of bills
delay presence, so every till would appear to go dark at exactly the
moment the system was busiest — the worst time to be blind to which
counters are alive.

Payloads are copied on the way in. paho reuses its buffer once a handler
returns and the work now happens after that, so a queued bill would
otherwise be read as whatever message arrived next — silently, and as
valid JSON often enough to commit the wrong sale.

One bug found by its own test: submit-after-stop selected between a
done-channel and the job channel, and once both were ready Go picks at
random. Picking the send panics on a closed channel. It would have shown
up in production as an occasional crash during shutdown and nowhere
else. Now guarded by an RWMutex held across the send, so the queue
cannot be closed under one in progress.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-03 19:53:39 +05:30
Suriya
1e3386fac8 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>
 2026-08-03 19:44:47 +05:30
Suriya
fc81df14e4 Answer the catalogue as a delta when the terminal sends a revision 
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>
 2026-08-03 19:26:01 +05:30
Suriya
8709556704 Add read endpoints for counter sales 
The ingest only ever wrote. A bill that reached pos_orders was safe and
completely unreachable — no screen in the product could show it, and the
only way to see a day's counter takings was to query the database by
hand.

Three endpoints: a paged bill list, one bill with its lines, and a
summary split the three ways somebody actually asks for — by tender for
reconciling a drawer, by day for a chart, by till for an outlet running
several counters.

locationid is required on all of them and is the authorisation boundary,
so a caller cannot page through another shop's takings by omitting a
parameter. Fetching a bill under the wrong outlet returns 404 even when
the reference is a real one.

Dates match businessdate rather than arrival, because a till that was
offline overnight uploads yesterday's bills this morning and they belong
to yesterday. The list is ordered by billedat for the same reason —
sorting by arrival would interleave a recovered backlog through today.

Unlike the ingest handlers these answer in the usual envelope: they are
read by the web app, not by a terminal, and nothing about them is bound
to the till's contract.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-03 19:17:03 +05:30
Suriya
ef647d3395 Give the POS its own broker accounts 
Two scoped users, pos_terminal and pos_ingest, with an ACL that keeps a
till to its own topics: it can publish its bills, registrations and
heartbeats, read its own acks, and nothing else. It cannot reach
nearle/riders/# or doormile/#, and cannot forge an ack — only the ingest
writes those.

admin is deliberately left unrestricted. Its credentials are compiled
into the rider app, so narrowing it would cut off the live fleet without
warning; that change needs someone to confirm nothing else uses it
first. Because admin's entry grants everything, applying the ACL changed
nothing for existing traffic — verified by watching riders 852 and 1114
keep publishing battery, speed and periodic logs throughout.

The scoping was verified by publishing as pos_terminal to four topics
and observing which arrived: the order did, the rider topic, the
doormile topic and its own ack topic did not.

Config and password file were backed up first; the rollback is one cp
and a container restart.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-03 18:01:05 +05:30
Suriya
64a219e7da Stop tracking .env 
It has been in the repository since the initial commit carrying the live
database host, user and password. Removing it from the index stops that
getting worse; the credentials are still in history and should be
rotated, which needs coordinating with everything that reads them.

Deployments should pass configuration as container environment rather
than shipping a file — a file on disk is one `git add -f` away from
being committed again.

Also closes the last untested path: a shopper registration published
over the broker rather than posted over HTTP. All three MQTT topics —
order, customer and health — have now been fired against the live
Mosquitto instance and acknowledged.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
 2026-08-03 17:56:09 +05:30
27 changed files with 2885 additions and 153 deletions
Expand all files Collapse all files

21
.env
View File

@@ -1,21 +0,0 @@
APP_PORT=1009
DB_HOST=66.116.207.225
DB_PORT=5433
DB_NAME=nearledb
DB_USER=admin
DB_PASSWORD="Package@123#"
# --- Catalogue Postgres / pgvector (separate DB, read-only integration) ---
CATALOGUE_DB_HOST=31.97.228.132
CATALOGUE_DB_PORT=6054
CATALOGUE_DB_NAME=pgvector
CATALOGUE_DB_USER=admin
CATALOGUE_DB_PASSWORD="'Package@321#'"
# --- DigitalOcean Spaces (S3-compatible), catalogue product images ---
USE_S3=true
S3_ACCESS_KEY=DO801G8Q8JAZKF49U3WJ
S3_SECRET_KEY=lBQExYfkVqH+ybmGVmQH5MkThBbrIohA/VQLgcPUvug
S3_ENDPOINT=https://nearle.sgp1.digitaloceanspaces.com
S3_BUCKET=nearle
S3_REGION=sgp1

8
.gitignore vendored
View File

@@ -48,3 +48,11 @@ Thumbs.db
*.mov
*.wmv
# Local configuration. Tracked until 2026-08-03, which put the database
# credentials in this repository's history — removing it from the index stops
# that getting worse, but the existing history still has them and the password
# should be rotated.
.env
.env.*
!.env.example

458
POS_API.md Normal file
View File

@@ -0,0 +1,458 @@
# POS integration — handover
Everything a developer needs to work on, extend or debug the in-store POS
integration. Companion to [`POS_TERMINAL_INGEST.md`](POS_TERMINAL_INGEST.md),
which covers deployment and broker setup.
Base path for everything below: **`/live/api/v1/pos`**
---
## 1. What this is
Retail tills run a Flutter POS app. Each one holds its own SQLite database and
keeps working with no network at all. When a connection is available it
publishes completed bills to an MQTT broker; a consumer in this backend commits
them to Postgres and acknowledges.
```
Cashier completes sale
│
▼
Till's SQLite ───────────────────── one transaction, before any network
sync_status = 0 survives crash, power cut, dead wifi
│
▼
MQTT broker ──────────────────────── transit only, holds nothing you can rely on
nearle/pos/{loc}/{terminal}/order
│
▼
fiesta consumer (messaging/posmqtt.go)
│
▼
POSTGRES ─────────────────────────── the permanent record
pos_orders, pos_order_items
productstocks (stock deducted here)
│
▼
ack → nearle/pos/{loc}/{terminal}/ack
│
▼
Till marks it synced, keeps its copy 7 more days, then purges
```
**Health** takes a separate path: every till publishes a heartbeat every 30
seconds, which lands in **Redis** under a 90-second TTL. Never in Postgres.
---
## 2. The four rules everything rests on
Break any of these and shops lose money. They are not stylistic.
**1. Only an application acknowledgement counts.**
A broker PUBACK means "I hold these bytes". It is not evidence the database
accepted anything. The ack is published *after* the transaction commits, never
from a handler that has merely queued the work.
**2. Silence is not acceptance.**
No ack, an empty ack, a 200 with no body — all leave the bill on the till, and
it is sent again. This is the correct behaviour when we are struggling.
**3. A duplicate is a success.**
Delivery is at-least-once. A lost ack makes a terminal re-send bills we already
hold. Reporting those as failures would strand a day of takings. Deduplication
is a unique index on `pos_orders.terminalorderid` — the UUID minted at the till
— plus a Postgres advisory lock held for the transaction.
**4. Store and terminal come from the topic, never the body.**
A till that could name its own store in a payload could redirect another
counter's acknowledgements.
---
## 3. Endpoints written by a terminal
These answer with a **bare body**, not the usual `{code, message, status}`
envelope — the till reads `accepted` from the top level and marks a bill synced
only if its id is there. Wrapping it would leave every terminal queueing for
ever.
### `POST /orders`
```json
{
"schema": 1,
"batch_id": "9f1c…",
"store_id": "1135",
"terminal_id": "T4A9",
"orders": [{
"id": "99999999-8888-4777-8666-555555555555",
"invoice_number": "INV-2608-T4A9-00002",
"created_at": "2026-08-03T17:29:00Z",
"cashier": "Divya",
"customer": {"id": "…", "mobile": "9840099999", "name": "Ravi"},
"subtotal": 60.0, "discount": 0.0, "tax": 4.44,
"tax_breakdown": {"0.08": 4.44},
"round_off": 0.0, "total": 60.0,
"points_earned": 0, "points_redeemed": 0,
"payments": [{"method": "upi", "amount": 60.0, "reference": "TXN123"}],
"items": [{
"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
}]
}]
}
```
Response:
```json
{ "batch_id": "9f1c…", "accepted": ["99999999-…"], "rejected": {} }
```
Status codes carry the other half of the contract:
| Code | Meaning | Terminal does |
|---|---|---|
| `200` | batch processed; ack says which bills landed | marks the named ids synced |
| `4xx` | the request is wrong — unknown outlet, bad store id | **halts** and shows a person |
| `5xx` | outcome unknown | keeps everything, retries with backoff |
### `POST /customers`
Same envelope with a `customers` array. **Insert-if-absent on id** — never an
update, so a profile corrected at head office is not reverted by a terminal
replaying an old capture.
The id is a **UUIDv5 over the normalised ten-digit mobile**, so two tills
registering the same shopper independently produce the same row. Do not
reassign it.
No loyalty figures travel upward — points and spend are derived from the bill
stream, which is idempotent and sees every counter.
---
## 4. GET endpoints — for the web app
**These use the normal `{code, message, status, details}` envelope.**
### `GET /sales` — bills for an outlet
```bash
curl "$BASE/sales?locationid=1135&fromdate=2026-08-01&todate=2026-08-03"
```
| Parameter | |
|---|---|
| `locationid` | **required** — the authorisation boundary |
| `fromdate`, `todate` | `YYYY-MM-DD`, matched on `businessdate` |
| `terminalid` | e.g. `T4A9` |
| `cashiername` | exact match |
| `paymentmode` | `cash`, `card`, `upi`, `wallet` |
| `pageno` | 0-based, default 0 |
| `pagesize` | default 50, max 500 |
```json
{
"code": 200, "status": true,
"details": {
"total": 137, "pageno": 0, "pagesize": 50,
"bills": [{
"posorderid": 7,
"terminalorderid": "99999999-8888-4777-8666-555555555555",
"invoicenumber": "INV-2608-T4A9-00002",
"tenantid": 1087, "locationid": 1135,
"terminalid": "T4A9", "cashiername": "Divya",
"customerid": 6847, "customermobile": "9840099999", "customername": "Ravi",
"billedat": "2026-08-03T17:29:00Z",
"businessdate": "2026-08-03",
"subtotal": 60, "discount": 0, "taxamount": 4.44,
"roundoff": 0, "total": 60,
"pointsearned": 0, "pointsredeemed": 0,
"itemcount": 1, "paymentmode": "upi",
"paymentsjson": "[{\"method\":\"upi\",\"amount\":60,\"reference\":\"TXN123\"}]",
"promosjson": "[]",
"taxbreakdownjson": "{\"0.08\":4.44}",
"batchid": "batch-mqtt-0001",
"receivedat": "2026-08-03T17:29:11Z"
}]
}
}
```
Line items are **not** included — a page of 50 bills would drag hundreds of rows
behind it and a list screen shows none of them. Use `/sales/detail`.
Ordered by `billedat` descending, not by id: a backlog uploaded after an outage
arrives out of order, and sorting by arrival would interleave yesterday's bills
through today's.
### `GET /sales/detail` — one bill with its lines
```bash
curl "$BASE/sales/detail?locationid=1135&reference=INV-2608-T4A9-00002"
```
`reference` accepts **any of three**: the terminal's order UUID, the invoice
number, or the `posorderid`. A support call starts from whichever the caller
happens to be looking at.
```json
{
"code": 200, "status": true,
"details": {
"posorderid": 7,
"invoicenumber": "INV-2608-T4A9-00002",
"…": "all the fields above, plus:",
"items": [{
"posorderitemid": 12, "posorderid": 7,
"productid": 6988, "productname": "Mysore Banana",
"barcode": "6988", "unitname": "kg",
"quantity": 1, "unitprice": 60,
"discountamount": 0, "gstrate": 0.08,
"taxamount": 4.44, "linetotal": 60
}]
}
}
```
Returns **404** if the reference does not belong to that `locationid` — even
when the reference is a real bill at another outlet.
### `GET /sales/summary` — totals
```bash
curl "$BASE/sales/summary?locationid=1135&fromdate=2026-08-01&todate=2026-08-03"
```
Takes the same filters as `/sales`.
```json
{
"code": 200, "status": true,
"details": {
"locationid": 1135,
"fromdate": "2026-08-01", "todate": "2026-08-03",
"billcount": 137, "itemcount": 402,
"grosssales": 18450.50, "taxcollected": 1204.30,
"discountgiven": 320.00, "roundoff": -1.50,
"averagebill": 134.68,
"bypaymentmode": [
{"paymentmode": "cash", "billcount": 80, "amount": 9200.00},
{"paymentmode": "upi", "billcount": 57, "amount": 9250.50}
],
"byday": [
{"businessdate": "2026-08-01", "billcount": 44, "amount": 5900.00}
],
"byterminal": [
{"terminalid": "T4A9", "billcount": 137, "amount": 18450.50}
]
}
}
```
Three breakdowns because they answer three different questions: **by tender**
for reconciling a drawer, **by day** for a chart, **by till** for an outlet
running several counters.
### `GET /health/terminal` — one till
```bash
curl "$BASE/health/terminal?terminal_id=T4A9"
```
```json
{
"code": 200, "status": true,
"details": {
"terminal_id": "T4A9", "location_id": "1135",
"store_name": "Ragul stores Selvapuram",
"app_version": "1.1.0", "status": "online",
"pending_bills": "0", "pending_registrations": "0",
"oldest_pending_at": "",
"today_bills": "2", "today_amount": "170",
"last_bill_at": "",
"printer_reachable": "0",
"reported_at": "2026-08-03T12:04:21Z",
"received_at": "2026-08-03T12:04:21Z"
}
}
```
Values are **strings** — it is a Redis hash. A till that has not reported inside
its TTL returns `200` with `status: "offline"`, not a 404: it exists, it is
simply quiet.
Fields the till does not collect are **absent, not zero**. A board showing every
terminal at 0% battery is worse than one showing nothing.
`pending_bills` is the number worth watching. A shop quietly accumulating
unsynced takings looks completely normal from the floor.
### `GET /health/location` — the "which counters are dark" board
```bash
curl "$BASE/health/location?location_id=1135"
```
```json
{
"code": 200, "status": true,
"details": {
"location_id": "1135", "total": 3, "online": 2,
"terminals": [
{"terminal_id": "T4A9", "status": "online", "today_bills": "37", "…": "…"},
{"terminal_id": "T7B2", "status": "offline", "reason": "no heartbeat within 90s"}
]
}
}
```
A till whose key expired comes back marked **offline rather than omitted** —
omitting it would make a dead terminal indistinguishable from one that was never
installed, and the dead one is exactly what somebody is looking for.
### `GET /catalogue` — the till's product pull
Bare body, no envelope. Used by terminals, not the web app.
```bash
curl "$BASE/catalogue?store_id=1135&page_size=500"
curl "$BASE/catalogue?store_id=1135&since=loc1135-20260803T135407Z"
```
No `since` → **full snapshot**. With a valid `since` → **change set**.
```json
{
"revision": "loc1135-20260803T135407Z",
"is_delta": false,
"has_more": false,
"products": [{
"id": "6988", "name": "Mysore Banana",
"barcode": "6988", "sku": "",
"category": "grocery", "price": 60, "stock": 750,
"unit": "kilogram", "gst_rate": 0.08, "is_active": true
}],
"customers": [],
"retired_product_ids": []
}
```
**`is_delta` is the dangerous field.** `false` means the terminal withdraws
every product the response does not mention. A filtered result labelled `false`
empties the shelf. In `Catalogue()` the filter and the flag are derived from one
value, so no code path can set one without the other.
Anything ambiguous resolves toward the snapshot: a revision that is malformed,
empty, or issued to a different outlet yields a full response.
The revision **only advances on the final page**, so a terminal that abandons a
paginated pull cannot end up holding one claiming it saw pages it never got.
A delta **cannot withdraw a deleted product** — removing a row from
`productlocations` leaves no tombstone. Only a snapshot collects those, so tills
should pull without a revision periodically.
---
## 5. MQTT topics
| Topic | Direction | Retained |
|---|---|---|
| `nearle/pos/{loc}/{terminal}/order` | till → us | no |
| `nearle/pos/{loc}/{terminal}/customer` | till → us | no |
| `nearle/pos/{loc}/{terminal}/health` | till → us, 30s | no |
| `nearle/pos/{loc}/{terminal}/ack` | us → till | no |
| `nearle/pos/{loc}/{terminal}/status` | till → us | **yes** (Last Will) |
| `nearle/pos/{loc}/catalogue` | us → all tills at a shop | **yes** |
`{loc}` is the numeric `tenantlocations.locationid`. The tenant is resolved from
it server-side and never taken from the wire.
Namespaced under `nearle/` alongside the rider fleet's `nearle/riders/…`.
---
## 6. Database
**`pos_orders`** — one row per counter bill. Separate from `orders` because a
bill carries a cashier, terminal, rounding, promos, loyalty and a payment split
that `orders` has nowhere to put.
**`pos_order_items`** — one row per line.
**`productstocks`** — stock is **not** separate. A counter sale writes the same
`out` rows an app order does, through helpers in `repositories/stockLedger.go`.
Two stock ledgers would mean the catalogue pull sends a till figures that ignore
its own trading.
**`customers`** — registrations, matched on `contactno`.
**Redis** — `pos:terminal:{code}` (hash, 90s TTL) and
`pos:location:{id}:terminals` (set, no TTL). Namespaced `pos:*` so they cannot
collide with express's `delivery:*`, `city:*`, `rider_*`.
> **Reporting:** counter sales are unioned into `GetRevenueSummary` and
> `GetSalesSummary`. **Any new report must do the same**, or it will silently
> understate every shop that runs a till. That is the standing cost of the split.
---
## 7. Code map
| File | |
|---|---|
| `models/pos.go` | wire types — matches the till's JSON exactly |
| `models/posorder.go` | `pos_orders` / `pos_order_items` |
| `models/poshealth.go` | heartbeat |
| `repositories/posRepository.go` | ingest + catalogue |
| `repositories/posSalesRepository.go` | the GET reads |
| `repositories/posPresence.go` | Redis presence |
| `repositories/stockLedger.go` | **shared** stock helpers |
| `messaging/posmqtt.go` | MQTT consumer |
| `messaging/posworkers.go` | bounded worker pools |
| `controllers/posController.go` | HTTP handlers |
| `routes/posroutes.go` | routes |
---
## 8. Operational notes
**Only `fiesta-0` consumes.** MQTT has no queue groups, so all replicas would
receive every message and commit the same bill three times. Ordinal 0 is
elected; the others log *"not the elected consumer"*. Override with
`POS_MQTT_CONSUMER=always|never`.
**Worker pools**: `POS_INGEST_WORKERS` (default 8), `POS_HEALTH_WORKERS`
(default 2). Heartbeats have their own pool so a backlog of bills cannot make
every till look dark at the busiest moment. A full queue **blocks**, pushing
backpressure to the broker and the till — slow, never lossy.
**Startup check**: three `pos: subscribed to nearle/pos/+/+/…` lines on
`fiesta-0`. Without them, MQTT ingest is not running and tills queue silently.
**The broker is not durable storage.** Mosquitto's `max_queued_messages` is 1000
and it flushes every 30 minutes. Fine, because a till keeps its copy until we
acknowledge — but nobody may ever ack on the broker's behalf.
---
## 9. Known gaps
- **No product prices.** Every product at loc 1135 is ₹0, so nothing is
sellable. Unpriced products come down as `is_active: false` so a till cannot
ring up a ₹0 item.
- **The Flutter app has never been run.** All testing used a Go program
impersonating a till.
- **No TLS** on port 1883. Bills carry customer names and mobile numbers.
- **No device authentication.** A terminal is trusted with a location id.
- **Loyalty does not come back down.** Balances at a till are that till's view.
- **`productstocks.quantity` is an integer** but tills sell in kg. POS rounds
**up** so it never under-deducts; the app-order path truncates, which was left
alone rather than silently changed. Making the column numeric is the real fix.
- **Broker credentials in source** — `admin` is in the rider APK and still
unrestricted. The two POS accounts are the only ones not in a source tree.

167
POS_TERMINAL_INGEST.md
View File

@@ -52,11 +52,41 @@ the standing cost of the split.
### HTTP
Base path: `/live/api/v1/pos`
**Written by a terminal** — bare-ack responses, see below.
| Method | Path | Purpose |
|---|---|---|
| `POST` | `/live/api/v1/pos/orders` | Completed bills |
| `POST` | `/live/api/v1/pos/customers` | Shoppers registered at a till |
| `GET` | `/live/api/v1/pos/catalogue` | Product pull. Query: `store_id`, `since`, `page`, `page_size` |
| `POST` | `/orders` | Completed bills |
| `POST` | `/customers` | Shoppers registered at a till |
| `GET` | `/catalogue` | Product pull. Query: `store_id`, `since`, `page`, `page_size` |
**Read by the web app** — normal `{code, message, status, details}` envelope.
| Method | Path | Purpose |
|---|---|---|
| `GET` | `/sales` | Bills for an outlet, newest first |
| `GET` | `/sales/detail` | One bill with its lines |
| `GET` | `/sales/summary` | Totals by tender, day and till |
| `GET` | `/health/terminal` | One till's live state |
| `GET` | `/health/location` | Every till at a shop |
`/sales` and `/sales/summary` take: **`locationid` (required)**, `fromdate`,
`todate` (YYYY-MM-DD, matched on `businessdate`), `terminalid`, `cashiername`,
`paymentmode`, `pageno`, `pagesize`.
`/sales/detail` takes `locationid` and `reference` — the terminal's order UUID,
the invoice number, or the `posorderid`, whichever the caller happens to have.
**`locationid` is the authorisation boundary.** Every read is scoped to one
outlet; omitting it is an error rather than a page through every shop's takings,
and asking for a bill under the wrong outlet returns 404 even when the reference
is valid.
Dates match `businessdate` — the day the sale was rung, not the day it reached
us. A till that was offline overnight uploads yesterday's bills this morning and
they belong to yesterday.
These answer with a **bare ack**, not the usual `{code, message, status}`
envelope — the terminal reads `accepted` from the top level of the body:
@@ -162,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
@@ -225,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
@@ -239,75 +299,56 @@ state.
as zero: a board showing every till at 0% battery is worse than one showing
nothing.
## Broker hardening — before a hundred tills join
## Broker accounts
Measured on the live broker, not assumed. None of this is caused by the POS
work; all of it gets worse the moment bills start flowing.
Applied 2026-08-03 on `66.116.225.226`. Two scoped accounts now exist alongside
`admin`, with an ACL at `/mosquitto/config/acl` referenced from
`mosquitto.conf`.
**There is no ACL file.** `allow_anonymous false` is set and auth is by password
file, but with no `acl_file` every authenticated user is unrestricted on every
topic. The rider app ships `admin` credentials **hardcoded in its APK**, so
anyone who decompiles it today has full publish and subscribe over `nearle/#`
*and* `doormile/#` — a second project's traffic. Adding POS puts every shop's
takings behind the same credential.
| User | May publish | May subscribe |
|---|---|---|
| `pos_terminal` | `nearle/pos/+/+/{order,customer,status,health}` | `nearle/pos/+/+/{ack,command}`, `nearle/pos/+/catalogue` |
| `pos_ingest` | `nearle/pos/+/+/{ack,command}`, `nearle/pos/+/catalogue` | `nearle/pos/+/+/{order,customer,health,status}` |
| `admin` | everything — **deliberately unchanged** | everything |
A scoped account is two commands and a container restart:
A till therefore cannot publish to `nearle/riders/#` or `doormile/#`, and cannot
write its own ack topic — only the ingest may do that. Verified by publishing as
`pos_terminal` to all four and watching which arrived: the order did, the other
three did not.
```bash
# A user for the tills, and one for this backend.
mosquitto_passwd -b /mosquitto/config/passwd pos_terminal '<strong-unique-pw>'
mosquitto_passwd -b /mosquitto/config/passwd pos_ingest '<different-pw>'
```
**`admin` was left unrestricted on purpose.** Its credentials are compiled into
the rider app, so narrowing it here would cut off the live rider fleet without
warning. The right next step is:
```conf
# /mosquitto/config/acl — then add `acl_file /mosquitto/config/acl` to mosquitto.conf
# Tills: publish their own traffic, read only their own acks and their shop's
# catalogue. The %c substitution binds a client to its own topics, so one till
# cannot read another's.
user pos_terminal
topic write nearle/pos/+/+/order
topic write nearle/pos/+/+/customer
topic write nearle/pos/+/+/status
topic write nearle/pos/+/+/health
topic read nearle/pos/+/+/ack
topic read nearle/pos/+/+/command
topic read nearle/pos/+/catalogue
# This backend: the mirror image.
user pos_ingest
topic read nearle/pos/+/+/order
topic read nearle/pos/+/+/customer
topic read nearle/pos/+/+/health
topic read nearle/pos/+/+/status
topic write nearle/pos/+/+/ack
topic write nearle/pos/+/+/command
topic write nearle/pos/+/catalogue
# Existing projects, scoped to what they already use.
user admin
topic readwrite nearle/riders/#
topic readwrite doormile/#
```
Tighten `pos_terminal` further with per-terminal credentials if you want one
till unable to read another's acks at all; the pattern above trusts tills within
the fleet but not outside it.
but only once someone has confirmed nothing else authenticates as `admin`.
Until then the ACL changes nothing for it — which is why applying it was safe.
**There is no TLS.** Port 8883 is not configured and is closed. Rider GPS
travels in the clear today; POS bills carry customer names and mobile numbers,
which is a different category of exposure on a shared network. Adding a listener
means certs plus republishing the port, i.e. recreating the container — worth
doing before rollout rather than after.
Rollback, if ever needed:
**Two more, from the audit:**
```bash
cp /root/Mqtt/backup-<timestamp>/{mosquitto.conf,passwd} /root/Mqtt/config/
docker restart mqtt_broker
```
- The broker password and the workolik NATS password differ only in
capitalisation. Diverge them when creating the scoped users.
- Confirm on the host whether the broker was started from the compose file or
from a bare `docker run` before editing the compose file and expecting it to
take effect — there is precedent in this estate for compose existing but not
being the deploy path.
**Still outstanding on the broker:**
- **No TLS.** Port 8883 is not configured. Bills carry customer names and mobile
numbers, and they travel in the clear. Traefik on the same host already
terminates 443, so certificates exist to borrow from.
- **`passwd` is world-readable.** Mosquitto warns about it and future versions
will refuse to load it. Tightening it means `chown 1883:1883` as well as
`chmod`, because the broker runs as uid 1883 and a root-owned 0600 file would
stop it starting.
- **Credentials in source.** `admin` is in the rider APK, Redis is hardcoded in
the express backend, and Postgres was in this repository's git history until
2026-08-03. The POS accounts above are the only ones not in any source tree —
keep it that way.
## Capacity

95
SECURITY_HANDOFF.md Normal file
View File

@@ -0,0 +1,95 @@
# Handoff: Broken Access Control (IDOR) audit & fixes — Fiesta backend
Repo: `backend_fiesta` (Go + Fiber + GORM), consumed by `nearledaily/daily_merchant_web` (React/TS) and a mobile app (not in this repo).
## 1. The root problem (still not fully fixed — read this first)
**There is no authentication system in this backend.** Grep confirms:
- No JWT/session token is ever issued. `Login`, `TenantLogin`, `TenantWebLogin`, `AppLogin` (in `controllers/userController.go`) just look up the user/tenant and return their info in the JSON body — no token.
- No auth middleware exists anywhere. `routes/routes.go` / `main.go` only wire up CORS middleware. Every route is wide open — anyone who can reach the API can call any endpoint with any query params.
Because of that, every endpoint trusts client-supplied query params (`tenantid`, `customerid`, `partnerid`, etc.) as the sole source of "who is asking." There is currently **nothing stopping a logged-in store admin for tenant 1135 from just requesting `?tenantid=1136`** and getting another tenant's data — the frontend happens to always send the logged-in user's own tenantid, but the backend never checks it.
**This session's fixes only close one specific hole**, described below. The real fix — deriving identity server-side from a verified token instead of trusting query params — has not been started. Whoever picks this up should treat that as the actual next milestone.
## 2. The specific bug that was found and fixed this session
Pattern found repeatedly across the codebase: repository functions build SQL dynamically, e.g.
```go
query := "SELECT ... FROM orders WHERE 1=1"
if tenantID != 0 {
query += " AND tenantid = ?"
params = append(params, tenantID)
}
// ...similar optional blocks for partnerid, customerid, etc.
```
**If none of the scoping params were supplied (0 / empty), the query silently fell through to "no WHERE clause" and returned every row in the table across every tenant.** This was directly reachable — e.g. `orders/getorders` with no `tenantid` returned all ~300 orders in the DB rather than 400ing, which is how the user first noticed this (logged in as a store admin, expected only their store's orders, saw everyone's).
### Fix pattern applied
Rather than rewriting every repository query (large surface area, higher regression risk), a **controller-level guard** was added to each affected endpoint: if none of the valid scoping ids are present in the query string, return `400` immediately instead of calling the service/repo at all.
Standard error shape used everywhere:
```json
{ "status": false, "code": 400, "message": "<specific message>" }
```
## 3. Endpoints fixed (8 total)
| # | Endpoint | File / function | Guard added |
|---|---|---|---|
| 1 | `GET /v1/web/orders/getorders` (+ mob) | `controllers/orderController.go` `GetOrders` (line 24) | requires one of `tenantid`, `partnerid`, `customerid`, `applocationid`, `appuserid` — else 400 (line ~102-110). Previously the `else` branch called `GetAllOrders` (unscoped). |
| 2 | `GET /v1/web/orders/getordersummary` | `controllers/orderController.go` `GetOrderSummary` (line 129) | requires one of `tenantid`, `partnerid`, `customerid`, `locationid` (line 137-143) |
| 3 | `GET /v1/web/orders/getlocationsummary` | `controllers/orderController.go` `GetlocationOrderSummary` (line 163) | requires `tenantid` (line 167-173) |
| 4 | `GET /v1/web/users/getallusers` | `controllers/userController.go` `GetAllUsers` (line 22) | requires `tenantid` (line 29-35). Note: this endpoint's query selects `a.pin` (login PIN) — this was a high-severity leak (PINs across all tenants) before the fix. |
| 5 | `GET /v1/web/deliveries/getdeliveries` (+ mob) | `controllers/deliveriesController.go` `GetDeliveries` (line 194) | requires one of `tenantid`, `partnerid`, `customerid`, `applocationid`, `userid`, `appuserid` (line 212-218) |
| 6 | `GET /v1/web/partners/getriders` (+ mob) | `controllers/partnerController.go` `GetActiveRiders` (line 19) | requires one of `tenantid`, `partnerid`, `applocationid`, `userid` (line 25-31). Lower severity — underlying repo query defaults to `userid = 0` rather than a full dump, but fixed for consistency. |
| 7 | `GET /v1/web/partners/getriderlogs` (+ mob) | `controllers/partnerController.go` `GetRiderLogs` (line 121) | requires one of `partnerid`, `applocationid` (line 127-133). **Also fixed an unrelated bug in the same function**: `tdate` was reading `c.Query("fromdate")` (copy-paste error) so the end of any date range was always silently overwritten with the start date. Now correctly reads `c.Query("todate")` (line 125). |
| 8 | `POST /v1/mob/orders/getcustomerorders` | `controllers/orderController.go` `GetCustomerOrders` (line 374) | requires `customerid` (line 394-400) |
### Also fixed alongside #2: SQL injection in `GetOrderSummary`
`repositories/orderRepository.go` `GetOrderSummary` previously built the date filter by **string-concatenating** `fdate`/`tdate` directly into raw SQL. Rewritten to use parameterized `?` placeholders passed through `r.db.Raw(query, params...)`. The `strconv` import was removed from that file since it became unused after the rewrite (verified via grep no other usage remained).
## 4. Reviewed and explicitly NOT changed (don't re-flag these)
Same `WHERE 1=1` pattern exists elsewhere but was judged not to be a bug, or already safe:
- **`repositories/tenantRepository.go` `GetAllTenants`** — intentionally lists all tenants for a platform/super-admin console. The gap here is "no RBAC to restrict who can call this," which is the same root-cause auth gap from section 1, not a scoping bug to patch individually.
- **`repositories/productRepository.go` `GetProductSubCategory`** — has an explanatory comment: subcategories are intentionally shared/global master data plus tenant-owned overrides. Not a bug.
- **`repositories/productRepository.go` `GetProductCount`** — returns aggregate counts only (no PII), low severity, left as-is.
- **`repositories/utilsRepository.go` `GetSubcategories`** — global taxonomy/reference data; the model has no tenant field at all.
- **`repositories/orderRepository.go` `GetAdminOrders`** — has `WHERE 1=1` internally but is safe because its only caller (`GetOrders` controller) only invokes it when `applocationid != 0`.
- **`repositories/orderRepository.go` `GetAllOrders`** — now dead code (unreachable) after fix #1 above; confirmed via grep it's no longer called anywhere. Could be deleted as cleanup but left in place.
- **`repositories/tenantRepository.go` `GetTenantLocations`** — already always filters `WHERE tenantid = ?`. Safe, unchanged.
## 5. Known pre-existing bug found during this audit, NOT yet fixed anywhere
**Frontend/backend path mismatch on rider logs.** In `nearledaily/daily_merchant_web/src/services/fiestaApi.ts`, `getRiderLogs()` (~line 1132) calls:
```ts
fiestaGet('riders/getriderlogs', {...})
```
`FIESTA_BASE` is `https://fiesta.nearle.app/live/api/v1/web`, so this resolves to `.../v1/web/riders/getriderlogs`. But the backend only registers this route under the `partners` group (`routes/partnerroutes.go`): `partner.Get("/getriderlogs", ...)` on `api.Group("/v1/web/partners")`, i.e. the real path is `.../v1/web/partners/getriderlogs`. Confirmed via grep there is no `/v1/web/riders` route group anywhere in the backend.
**This means `getRiderLogs()` in the web console has likely been 404ing already, independent of anything fixed this session.** Fix is a one-line FE change: `'riders/getriderlogs'` → `'partners/getriderlogs'`. Not fixed yet because it's a frontend-repo change and wasn't the scope of this backend security pass — flagging it here so it isn't lost.
## 6. Frontend compatibility check (already done, no FE changes needed for the 8 fixes above)
Checked `daily_merchant_web/src/services/fiestaApi.ts` against every fix — the web console already sends the now-required params in all cases:
- `getOrders`, `getAllUsers`, `getDeliveries`, `getOrderSummary`, `getLocationSummary` — all declare `tenantid: number` as a **required** (non-optional) TS field already.
- `getRiders` — always sends `applocationid: opts.applocationid ?? FIESTA_APPLOCATION_ID` (never zero/undefined) plus required `tenantid`.
- `getRiderLogs` — sends `tenantid`/`applocationid` when available, but see the path bug in section 5 — worth re-verifying once that's fixed.
- **`mob/orders/getcustomerorders`** (fix #8) is called from the **mobile app**, which is not in this repo — whoever owns that codebase needs to verify every call site always sends `customerid`. Not verified in this session.
## 7. Environment note
**No Go toolchain is available in the sandbox this session ran in** (`command not found: go`). All edits above were manually reviewed (imports, syntax, call sites checked via Read/grep) but **never compiled**. Run `go build ./...` and the existing test suite (if any) before deploying any of this.
## 8. Suggested next steps for whoever picks this up
1. `go build ./...` and smoke-test all 8 changed endpoints (call with and without the required param, confirm 200 vs 400).
2. Fix the `riders/getriderlogs` → `partners/getriderlogs` path bug in `fiestaApi.ts` (section 5).
3. Verify the mobile app always sends `customerid` to `mob/orders/getcustomerorders` before this ships, since that's the one fixed endpoint not verified from a frontend contract.
4. Scope and plan the real fix: JWT/session auth issuance + middleware, so `tenantid`/`customerid`/etc. are derived from a verified server-side identity instead of trusted from query params. Until that lands, the 8 fixes in this doc only prevent the "forgot to pass an id → get everything" failure mode — they do **not** prevent a malicious or buggy client from passing a *different* tenant's/customer's/partner's real id and getting their data.

159
controllers/posController.go
View File

@@ -1,6 +1,7 @@
package controllers
import (
"fmt"
"log"
"net/http"
"strconv"
@@ -94,6 +95,63 @@ func (ctl *PosController) IngestCustomers(c *fiber.Ctx) error {
return c.Status(http.StatusOK).JSON(ack)
}
// IngestHealth records one heartbeat from a till.
//
// The same heartbeat the broker carries, over HTTP, because presence was
// previously reachable *only* over MQTT — a terminal configured for the HTTP
// route reported bills perfectly and never appeared on the fleet board at all,
// with nothing anywhere to say why. A monitoring feature that silently does not
// exist on one of two supported transports is worse than no feature.
//
// Answers 202 rather than 200: nothing is committed, and the till is told not
// to wait on it. Failures are swallowed for the same reason the MQTT path
// swallows them — a terminal that cannot say how it is must still sell, and a
// blank square on a dashboard beats a till that stopped because Redis was busy.
func (ctl *PosController) IngestHealth(c *fiber.Ctx) error {
var health models.PosHealth
if err := c.BodyParser(&health); err != nil {
return c.Status(http.StatusBadRequest).JSON(fiber.Map{
"code": http.StatusBadRequest,
"message": "could not read the heartbeat: " + err.Error(),
"status": false,
})
}
// Over MQTT these come from the topic. There is no topic here, so the body
// is the only source and both are required — a heartbeat that cannot say
// which till it belongs to is unfilable.
if strings.TrimSpace(health.Terminalid) == "" {
return c.Status(http.StatusBadRequest).JSON(fiber.Map{
"code": http.StatusBadRequest, "status": false,
"message": "terminal_id is required",
})
}
if strings.TrimSpace(health.Locationid) == "" {
return c.Status(http.StatusBadRequest).JSON(fiber.Map{
"code": http.StatusBadRequest, "status": false,
"message": "location_id is required",
})
}
// Matches the consumer: a bare {"status":"offline"} is a Last Will and must
// survive as-is, but an unset status from a till that is plainly talking to
// us means online.
if strings.TrimSpace(health.Status) == "" {
health.Status = "online"
}
if err := ctl.posService.RecordHealth(c.Context(), health); err != nil {
// Logged, not returned. See above — the till must not slow down for it.
log.Printf("pos: could not record heartbeat from %s/%s over HTTP: %v",
health.Locationid, health.Terminalid, err)
}
return c.Status(http.StatusAccepted).JSON(fiber.Map{
"status": true, "code": http.StatusAccepted,
})
}
// Catalogue answers a terminal's product pull.
func (ctl *PosController) Catalogue(c *fiber.Ctx) error {
storeID := strings.TrimSpace(c.Query("store_id"))
@@ -188,6 +246,107 @@ func (ctl *PosController) LocationHealth(c *fiber.Ctx) error {
})
}
// ---------------------------------------------------------------- Sales reads
//
// Unlike the ingest handlers above, these answer in the usual
// `{code, message, status, details}` envelope — they are read by the web app,
// not by a terminal, and nothing about them is bound to the till's contract.
// posSalesFilter reads the shared query parameters.
func posSalesFilter(c *fiber.Ctx) (models.PosSalesFilter, error) {
locationID, err := strconv.Atoi(strings.TrimSpace(c.Query("locationid")))
if err != nil || locationID <= 0 {
return models.PosSalesFilter{}, fmt.Errorf("locationid is required")
}
pageno, _ := strconv.Atoi(c.Query("pageno", "0"))
pagesize, _ := strconv.Atoi(c.Query("pagesize", "50"))
return models.PosSalesFilter{
Locationid: locationID,
Fromdate: strings.TrimSpace(c.Query("fromdate")),
Todate: strings.TrimSpace(c.Query("todate")),
Terminalid: strings.TrimSpace(c.Query("terminalid")),
Cashiername: strings.TrimSpace(c.Query("cashiername")),
Paymentmode: strings.TrimSpace(c.Query("paymentmode")),
Pageno: pageno,
Pagesize: pagesize,
}, nil
}
// GetSales lists counter bills for an outlet, newest first.
func (ctl *PosController) GetSales(c *fiber.Ctx) error {
filter, err := posSalesFilter(c)
if err != nil {
return posBadRequest(c, err)
}
page, err := ctl.posService.Sales(filter)
if err != nil {
return posServerError(c, "GetSales", err)
}
return c.JSON(fiber.Map{"code": http.StatusOK, "status": true, "details": page})
}
// GetSaleDetail returns one bill with its lines.
//
// Accepts the terminal's order UUID, the invoice number, or this backend's
// posorderid — a support call starts from whichever the caller is looking at.
func (ctl *PosController) GetSaleDetail(c *fiber.Ctx) error {
locationID, err := strconv.Atoi(strings.TrimSpace(c.Query("locationid")))
if err != nil || locationID <= 0 {
return posBadRequest(c, fmt.Errorf("locationid is required"))
}
reference := strings.TrimSpace(c.Query("reference"))
if reference == "" {
return posBadRequest(c, fmt.Errorf("reference is required — an order id, invoice number or posorderid"))
}
bill, err := ctl.posService.SaleDetail(locationID, reference)
if err != nil {
return posServerError(c, "GetSaleDetail", err)
}
if bill == nil {
return c.Status(http.StatusNotFound).JSON(fiber.Map{
"code": http.StatusNotFound,
"message": "no bill matches that reference at this outlet",
"status": false,
})
}
return c.JSON(fiber.Map{"code": http.StatusOK, "status": true, "details": bill})
}
// GetSalesSummary totals a range, split by tender, day and till.
func (ctl *PosController) GetSalesSummary(c *fiber.Ctx) error {
filter, err := posSalesFilter(c)
if err != nil {
return posBadRequest(c, err)
}
summary, err := ctl.posService.SalesSummary(filter)
if err != nil {
return posServerError(c, "GetSalesSummary", err)
}
return c.JSON(fiber.Map{"code": http.StatusOK, "status": true, "details": summary})
}
func posBadRequest(c *fiber.Ctx, err error) error {
return c.Status(http.StatusBadRequest).JSON(fiber.Map{
"code": http.StatusBadRequest, "message": err.Error(), "status": false,
})
}
func posServerError(c *fiber.Ctx, op string, err error) error {
log.Printf("pos %s: %v", op, err)
return c.Status(http.StatusInternalServerError).JSON(fiber.Map{
"code": http.StatusInternalServerError, "message": err.Error(), "status": false,
})
}
// posIngestError decides whether the terminal should retry.
//
// The distinction matters more than the message does. A misconfigured store id

107
messaging/posmqtt.go
View File

@@ -16,22 +16,17 @@ import (
mqtt "github.com/eclipse/paho.mqtt.golang"
)
// Plain-MQTT ingest for the Nearle POS terminals.
// MQTT ingest for the Nearle POS terminals.
//
// The sibling of posconsumer.go, and which one you want depends entirely on
// what is listening on the other end:
// The broker is Eclipse Mosquitto, shared with the rider fleet. An audit of the
// estate found no reachable NATS and no MQTT gateway on the NATS boxes that do
// exist, so a NATS consumer that briefly lived here was deleted rather than
// left to rot — a client for a protocol nothing speaks is worse than none.
//
// - **This file** talks MQTT to a broker like Mosquitto or EMQX — the kind
// already running at the rider app's `66.116.225.226:1883`.
// - **posconsumer.go** talks the NATS protocol to a NATS server, which
// exposes MQTT through a gateway but speaks NATS itself on 4222.
// Enabled with MQTT_URL. Unset, the terminals reach the same service over HTTP
// instead, and this file does nothing.
//
// They are not interchangeable: a NATS client cannot connect to Mosquitto, and
// an MQTT client cannot use NATS' native subjects. Both call the same
// PosService, so whichever is running, a bill lands identically.
//
// Enabled with MQTT_URL. Both may run at once, which is what a migration
// between brokers looks like.
// Only one replica consumes: see posConsumerElected.
const (
// Namespaced under `nearle/` alongside the rider app's
@@ -47,6 +42,12 @@ const (
type PosMqttConsumer struct {
client mqtt.Client
svc services.PosService
// Bills and registrations share a pool; heartbeats get their own, so a
// backlog of sales cannot make every till look dark at the moment the
// system is busiest.
ingest *posPool
health *posPool
}
// StartPosMqttConsumer connects and subscribes.
@@ -60,14 +61,47 @@ func StartPosMqttConsumer(svc services.PosService) (*PosMqttConsumer, error) {
return nil, nil
}
c := &PosMqttConsumer{svc: svc}
// Only one replica consumes.
//
// MQTT has no queue groups — every subscriber receives every message, so
// three replicas would each commit the same bill and publish three acks.
// The ingest is idempotent, so nothing double-counts, but it is three times
// the database work and three times the traffic for one sale.
//
// A StatefulSet gives pods stable ordinal names, so ordinal 0 is a
// deterministic election with no coordination and no extra dependency. If
// that pod dies the set recreates it; tills hold their bills and re-send in
// the meantime, which is exactly what they are built to do.
if !posConsumerElected() {
log.Printf("pos: replica %q is not the elected consumer, MQTT ingest idle here",
os.Getenv("HOSTNAME"))
return nil, nil
}
// Each ingest worker holds a database transaction while it runs, so the
// real ceiling is the Postgres connection pool rather than the CPU. The
// queue is deep enough to absorb a burst and shallow enough that a genuine
// overload is felt as backpressure rather than hidden as latency.
c := &PosMqttConsumer{
svc: svc,
ingest: newPosPool("ingest", posPoolSize("POS_INGEST_WORKERS", 8), 256),
health: newPosPool("health", posPoolSize("POS_HEALTH_WORKERS", 2), 512),
}
opts := mqtt.NewClientOptions().
AddBroker(url).
// Stable, so the broker resumes this session and redelivers anything
// in flight rather than treating every restart as a new subscriber.
SetClientID(getEnvDefault("MQTT_CLIENT_ID", "nearle-pos-ingest")).
// Defaults to the pod name so replicas can never collide: a second
// connection with the same client id evicts the first, and the two then
// fight in a reconnect loop that looks like a flapping network.
SetClientID(getEnvDefault("MQTT_CLIENT_ID",
getEnvDefault("HOSTNAME", "nearle-pos-ingest"))).
SetCleanSession(false).
// Ordered delivery keeps paho on one goroutine, which is what lets a
// full queue push back on the broker. With concurrent delivery paho
// would keep reading no matter how far behind the workers were.
SetOrderMatters(posOrderedDelivery).
SetAutoReconnect(true).
SetMaxReconnectInterval(30 * time.Second).
SetKeepAlive(30 * time.Second).
@@ -85,9 +119,9 @@ func StartPosMqttConsumer(svc services.PosService) (*PosMqttConsumer, error) {
opts.SetOnConnectHandler(func(client mqtt.Client) {
log.Printf("pos: connected to MQTT broker %s", url)
for topic, handler := range map[string]mqtt.MessageHandler{
topicOrders: c.handleOrders,
topicCustomers: c.handleCustomers,
topicHealth: c.handleHealth,
topicOrders: wrapHandler(c.ingest, c.handleOrders),
topicCustomers: wrapHandler(c.ingest, c.handleCustomers),
topicHealth: wrapHandler(c.health, c.handleHealth),
} {
if token := client.Subscribe(topic, 1, handler); token.Wait() && token.Error() != nil {
log.Printf("pos: could not subscribe to %s: %v", topic, token.Error())
@@ -254,6 +288,12 @@ func (c *PosMqttConsumer) Close() {
if c == nil || c.client == nil {
return
}
// Workers drain before the connection closes, so a bill mid-commit still
// gets its ack out. Disconnecting first would strand it: committed here,
// unacknowledged there, and sent again on the till's next attempt.
c.ingest.stop()
c.health.stop()
quiesce, err := strconv.Atoi(getEnvDefault("MQTT_QUIESCE_MS", "2000"))
if err != nil || quiesce < 0 {
quiesce = 2000
@@ -261,6 +301,37 @@ func (c *PosMqttConsumer) Close() {
c.client.Disconnect(uint(quiesce))
}
// posConsumerElected decides whether this replica runs the MQTT ingest.
//
// Rules, in order:
//
// - POS_MQTT_CONSUMER=always or =never settles it outright, for deployments
// that are not a StatefulSet or that want the consumer somewhere specific.
// - A StatefulSet pod name ending in `-0` is elected. Ordinals are stable and
// unique, so this needs no lock, no lease and no coordination.
// - Anything else — a bare container, a Deployment, local development —
// is elected, because a single instance that refused to consume would be a
// far more confusing failure than one that did.
func posConsumerElected() bool {
switch strings.ToLower(strings.TrimSpace(os.Getenv("POS_MQTT_CONSUMER"))) {
case "always", "true", "yes":
return true
case "never", "false", "no":
return false
}
host := strings.TrimSpace(os.Getenv("HOSTNAME"))
if i := strings.LastIndex(host, "-"); i >= 0 {
if ordinal := host[i+1:]; ordinal != "" && strings.Trim(ordinal, "0123456789") == "" {
// A StatefulSet ordinal. Only the first replica consumes.
return ordinal == "0"
}
}
// Not an ordinal-named pod, so there is nothing to elect against.
return true
}
func getEnvDefault(key, fallback string) string {
if v := strings.TrimSpace(os.Getenv(key)); v != "" {
return v

59
messaging/posmqtt_test.go
View File

@@ -56,6 +56,18 @@ func (f *fakePosService) TerminalHealth(context.Context, string) (map[string]str
return nil, nil
}
func (f *fakePosService) Sales(models.PosSalesFilter) (*models.PosSalesPage, error) {
return nil, nil
}
func (f *fakePosService) SaleDetail(int, string) (*models.PosOrders, error) {
return nil, nil
}
func (f *fakePosService) SalesSummary(models.PosSalesFilter) (*models.PosSalesSummary, error) {
return nil, nil
}
func (f *fakePosService) LocationHealth(context.Context, string) ([]map[string]string, error) {
return nil, nil
}
@@ -100,9 +112,9 @@ func (c *fakeClient) Subscribe(string, byte, mqtt.MessageHandler) mqtt.Token {
func (c *fakeClient) SubscribeMultiple(map[string]byte, mqtt.MessageHandler) mqtt.Token {
return doneToken{}
}
func (c *fakeClient) Unsubscribe(...string) mqtt.Token { return doneToken{} }
func (c *fakeClient) AddRoute(string, mqtt.MessageHandler) {}
func (c *fakeClient) OptionsReader() mqtt.ClientOptionsReader { return mqtt.ClientOptionsReader{} }
func (c *fakeClient) Unsubscribe(...string) mqtt.Token { return doneToken{} }
func (c *fakeClient) AddRoute(string, mqtt.MessageHandler) {}
func (c *fakeClient) OptionsReader() mqtt.ClientOptionsReader { return mqtt.ClientOptionsReader{} }
type doneToken struct{}
@@ -368,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)
}
})
}
}

174
messaging/posworkers.go Normal file
View File

@@ -0,0 +1,174 @@
package messaging
import (
"log"
"os"
"strconv"
"sync"
mqtt "github.com/eclipse/paho.mqtt.golang"
)
// Concurrency for the MQTT ingest.
//
// paho delivers messages on a single goroutine, so without this every bill is
// committed one after another. A bill is a full Postgres transaction — advisory
// lock, dedup check, stock row locks, availability check, four inserts, commit —
// which realistically costs 10–30ms. Serially that is 30–100 bills a second,
// and a shop-wide backlog draining after an outage would take minutes to land.
//
// ### Why a bounded pool rather than a goroutine per message
//
// paho can be told to call handlers concurrently, but it spawns without limit.
// A storm would then open a database transaction per message, exhaust the
// connection pool, and stall every one of them at once — turning a slow minute
// into a dead one.
//
// A fixed pool behind a bounded queue does the opposite. When the queue fills,
// submitting **blocks**, which is the point: paho stops acknowledging, the
// broker's in-flight window fills, and it stops sending. Backpressure travels
// all the way back to the till, which holds its bills and retries. Slow, but
// nothing is dropped and nothing is lost.
//
// ### Why bills and heartbeats have separate pools
//
// A heartbeat is one Redis write and a bill is a transaction. Sharing a queue
// would let a backlog of bills delay presence, and every till would appear to
// go dark at exactly the moment the system was busiest — the worst possible
// time to be blind to which counters are alive.
// posPool is a fixed set of workers reading a bounded queue.
type posPool struct {
name string
jobs chan func()
wg sync.WaitGroup
once sync.Once
// Guards the transition to closed. A plain `select` over a done-channel and
// the job channel is not enough: once both are ready Go picks between them
// at random, and picking the send panics on a closed channel. Held for
// reading across the whole of submit, so stop cannot close the queue out
// from under a send already in progress.
mu sync.RWMutex
closed bool
}
func newPosPool(name string, workers, queue int) *posPool {
p := &posPool{
name: name,
jobs: make(chan func(), queue),
}
p.wg.Add(workers)
for i := 0; i < workers; i++ {
go func() {
defer p.wg.Done()
for job := range p.jobs {
job()
}
}()
}
log.Printf("pos: %s pool started with %d workers, queue %d", name, workers, queue)
return p
}
// submit queues work, blocking when the queue is full.
//
// Blocking is deliberate. Dropping would lose a bill outright; the terminal
// would eventually re-send it, but only after its ack timeout, and meanwhile we
// would have thrown away work we had already accepted. Blocking instead pushes
// back through paho to the broker to the till, which is exactly where the
// decision to slow down belongs.
func (p *posPool) submit(job func()) {
p.mu.RLock()
if p.closed {
p.mu.RUnlock()
// Shutting down. Running it inline still gets the work done and its ack
// published, rather than discarding a bill that already reached us.
job()
return
}
// The read lock is held across the send. Blocking here while the queue is
// full cannot deadlock against stop: the workers only exit once the channel
// is closed, and that happens under the write lock this send is holding
// off — so they stay alive and keep draining until this send completes.
p.jobs <- job
p.mu.RUnlock()
}
// stop drains the queue and waits for in-flight work.
//
// Every job already accepted runs to completion, so a bill mid-commit still
// gets its ack. Without one the terminal would hold it and send it again on
// restart — harmless, but avoidable.
func (p *posPool) stop() {
p.once.Do(func() {
// The write lock waits for every submit already in progress, so the
// channel is never closed while something is mid-send.
p.mu.Lock()
p.closed = true
close(p.jobs)
p.mu.Unlock()
p.wg.Wait()
log.Printf("pos: %s pool drained", p.name)
})
}
// posPoolSize reads a worker count from the environment.
//
// The default is deliberately modest. Each worker holds a database transaction
// while it runs, so the useful ceiling is the Postgres connection pool, not the
// CPU — set this above what the database can serve and the workers simply queue
// inside the driver instead, where there is no backpressure to feel.
func posPoolSize(key string, fallback int) int {
v, err := strconv.Atoi(os.Getenv(key))
if err != nil || v <= 0 {
return fallback
}
if v > 128 {
return 128
}
return v
}
// posOrderedDelivery reports whether paho should preserve message order.
//
// Left on: paho then delivers on one goroutine, which hands work to the pool
// and blocks when it is full. That single delivery goroutine is what makes
// backpressure reach the broker at all — with concurrent delivery paho would
// keep reading regardless of how far behind the workers were.
const posOrderedDelivery = true
// wrapHandler puts a paho message handler behind a pool.
//
// The payload is copied because paho reuses its buffer once the handler
// returns, and the work now happens after that.
func wrapHandler(pool *posPool, h mqtt.MessageHandler) mqtt.MessageHandler {
return func(client mqtt.Client, msg mqtt.Message) {
topic := msg.Topic()
payload := make([]byte, len(msg.Payload()))
copy(payload, msg.Payload())
pool.submit(func() {
h(client, copiedMessage{topic: topic, payload: payload})
})
}
}
// copiedMessage carries a payload that outlives paho's buffer.
type copiedMessage struct {
topic string
payload []byte
}
func (m copiedMessage) Duplicate() bool { return false }
func (m copiedMessage) Qos() byte { return 1 }
func (m copiedMessage) Retained() bool { return false }
func (m copiedMessage) Topic() string { return m.topic }
func (m copiedMessage) MessageID() uint16 { return 0 }
func (m copiedMessage) Payload() []byte { return m.payload }
func (m copiedMessage) Ack() {}

247
messaging/posworkers_test.go Normal file
View File

@@ -0,0 +1,247 @@
package messaging
import (
"sync"
"sync/atomic"
"testing"
"time"
mqtt "github.com/eclipse/paho.mqtt.golang"
)
func TestEveryJobRuns(t *testing.T) {
pool := newPosPool("test", 4, 16)
var done int64
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
wg.Add(1)
pool.submit(func() {
defer wg.Done()
atomic.AddInt64(&done, 1)
})
}
wg.Wait()
pool.stop()
if got := atomic.LoadInt64(&done); got != 100 {
t.Errorf("ran %d jobs, want 100", got)
}
}
func TestConcurrencyIsBounded(t *testing.T) {
// The reason the pool exists. Unbounded concurrency would open a database
// transaction per message and exhaust the connection pool under a storm,
// stalling every one of them at once.
const workers = 4
pool := newPosPool("test", workers, 64)
var inFlight, peak int64
var wg sync.WaitGroup
for i := 0; i < 200; i++ {
wg.Add(1)
pool.submit(func() {
defer wg.Done()
now := atomic.AddInt64(&inFlight, 1)
for {
was := atomic.LoadInt64(&peak)
if now <= was || atomic.CompareAndSwapInt64(&peak, was, now) {
break
}
}
time.Sleep(time.Millisecond)
atomic.AddInt64(&inFlight, -1)
})
}
wg.Wait()
pool.stop()
if got := atomic.LoadInt64(&peak); got > workers {
t.Errorf("peak concurrency %d exceeded the %d workers", got, workers)
}
}
func TestSubmitBlocksRatherThanDroppingWork(t *testing.T) {
// A full queue must slow the caller down, not discard a bill. Dropping
// would throw away work already accepted from the broker, and the terminal
// would only find out at its ack timeout.
pool := newPosPool("test", 1, 1)
release := make(chan struct{})
var ran int64
// Occupy the single worker.
pool.submit(func() {
<-release
atomic.AddInt64(&ran, 1)
})
// Fill the queue, then a third submit must block until the worker frees up.
pool.submit(func() { atomic.AddInt64(&ran, 1) })
blocked := make(chan struct{})
go func() {
pool.submit(func() { atomic.AddInt64(&ran, 1) })
close(blocked)
}()
select {
case <-blocked:
t.Fatal("submit returned while the queue was full; work would be dropped under load")
case <-time.After(100 * time.Millisecond):
// Correctly blocked.
}
close(release)
select {
case <-blocked:
case <-time.After(3 * time.Second):
t.Fatal("submit never unblocked after the worker freed up")
}
pool.stop()
if got := atomic.LoadInt64(&ran); got != 3 {
t.Errorf("ran %d jobs, want 3 — none may be lost", got)
}
}
func TestStopDrainsAcceptedWork(t *testing.T) {
// A bill mid-commit must still get its ack. Without one the terminal holds
// it and sends it again on restart — harmless, but avoidable.
pool := newPosPool("test", 2, 64)
var done int64
for i := 0; i < 50; i++ {
pool.submit(func() {
time.Sleep(time.Millisecond)
atomic.AddInt64(&done, 1)
})
}
pool.stop()
if got := atomic.LoadInt64(&done); got != 50 {
t.Errorf("only %d of 50 jobs completed before shutdown finished", got)
}
}
func TestSubmitAfterStopStillRunsTheWork(t *testing.T) {
// A message that arrived during shutdown has already been taken from the
// broker. Discarding it would lose a bill we accepted responsibility for.
pool := newPosPool("test", 2, 8)
pool.stop()
var ran int64
pool.submit(func() { atomic.AddInt64(&ran, 1) })
if got := atomic.LoadInt64(&ran); got != 1 {
t.Error("work submitted during shutdown was discarded")
}
}
func TestStopIsIdempotent(t *testing.T) {
// Close() may be reached twice on a shutdown path; a second close of the
// jobs channel would panic and take the process down mid-drain.
pool := newPosPool("test", 2, 8)
pool.stop()
pool.stop()
pool.stop()
}
func TestPoolSizeFallsBackAndClamps(t *testing.T) {
t.Setenv("POS_TEST_WORKERS", "")
if got := posPoolSize("POS_TEST_WORKERS", 8); got != 8 {
t.Errorf("unset = %d, want the fallback 8", got)
}
t.Setenv("POS_TEST_WORKERS", "not a number")
if got := posPoolSize("POS_TEST_WORKERS", 8); got != 8 {
t.Errorf("garbage = %d, want the fallback 8", got)
}
t.Setenv("POS_TEST_WORKERS", "0")
if got := posPoolSize("POS_TEST_WORKERS", 8); got != 8 {
t.Errorf("zero = %d, want the fallback 8", got)
}
t.Setenv("POS_TEST_WORKERS", "-4")
if got := posPoolSize("POS_TEST_WORKERS", 8); got != 8 {
t.Errorf("negative = %d, want the fallback 8", got)
}
t.Setenv("POS_TEST_WORKERS", "24")
if got := posPoolSize("POS_TEST_WORKERS", 8); got != 24 {
t.Errorf("explicit = %d, want 24", got)
}
// Clamped: more workers than the database can serve just moves the queue
// inside the driver, where there is no backpressure to feel.
t.Setenv("POS_TEST_WORKERS", "100000")
if got := posPoolSize("POS_TEST_WORKERS", 8); got != 128 {
t.Errorf("absurd = %d, want the 128 clamp", got)
}
}
func TestAWrappedHandlerCopiesThePayload(t *testing.T) {
// paho reuses its buffer once a handler returns, and with a pool the work
// now happens *after* that. Without a copy a queued bill would be read as
// whatever message happened to arrive next — silently, and as valid JSON
// often enough to commit the wrong sale.
pool := newPosPool("test", 1, 4)
seen := make(chan string, 1)
wrapped := wrapHandler(pool, func(_ mqtt.Client, msg mqtt.Message) {
seen <- string(msg.Payload())
})
// A buffer paho would reuse.
buffer := []byte(`{"batch_id":"original"}`)
wrapped(nil, fakeMessage{topic: "nearle/pos/12/T4A9/order", payload: buffer})
// Overwrite it the instant the handler returns, exactly as paho would.
for i := range buffer {
buffer[i] = 'X'
}
select {
case got := <-seen:
if got != `{"batch_id":"original"}` {
t.Errorf("handler saw %q — the payload was not copied before queueing", got)
}
case <-time.After(3 * time.Second):
t.Fatal("the wrapped handler never ran")
}
pool.stop()
}
func TestAWrappedHandlerKeepsTheTopic(t *testing.T) {
// Store and terminal are read from the topic, never the body. Losing it in
// the hand-off would leave the ack with nowhere to go.
pool := newPosPool("test", 1, 4)
seen := make(chan string, 1)
wrapped := wrapHandler(pool, func(_ mqtt.Client, msg mqtt.Message) {
seen <- msg.Topic()
})
wrapped(nil, fakeMessage{topic: "nearle/pos/1135/T4A9/order", payload: []byte("{}")})
select {
case got := <-seen:
if got != "nearle/pos/1135/T4A9/order" {
t.Errorf("topic = %q, want nearle/pos/1135/T4A9/order", got)
}
case <-time.After(3 * time.Second):
t.Fatal("the wrapped handler never ran")
}
pool.stop()
}

71
models/posorder.go
View File

@@ -136,3 +136,74 @@ type PosOrderItems struct {
func (PosOrderItems) TableName() string {
return "pos_order_items"
}
// PosSalesFilter scopes a query over counter sales.
//
// Locationid is required and is the authorisation boundary — every read is
// scoped to one outlet, so a caller cannot page through another shop's takings
// by omitting a parameter.
type PosSalesFilter struct {
Locationid int
Fromdate string // YYYY-MM-DD, matched against businessdate
Todate string
Terminalid string
Cashiername string
Paymentmode string
Pageno int
Pagesize int
}
// PosSalesPage is one page of bills, with the total so a caller can paginate
// without a second request.
type PosSalesPage struct {
Total int64 `json:"total"`
Pageno int `json:"pageno"`
Pagesize int `json:"pagesize"`
Bills []PosOrders `json:"bills"`
}
// PosSalesSummary totals a range of counter sales.
//
// Deliberately separate from the bill list: a shop settling a till wants the
// figures, not five hundred rows, and computing them client-side would mean
// fetching every page first.
type PosSalesSummary struct {
Locationid int `json:"locationid"`
Fromdate string `json:"fromdate"`
Todate string `json:"todate"`
Billcount int `json:"billcount"`
Itemcount int `json:"itemcount"`
Grosssales float64 `json:"grosssales"`
Taxcollected float64 `json:"taxcollected"`
Discount float64 `json:"discountgiven"`
Roundoff float64 `json:"roundoff"`
Averagebill float64 `json:"averagebill"`
// What a cashier reconciles the drawer against.
Bypaymentmode []PosPaymentTotal `json:"bypaymentmode"`
// One row per trading day, for a chart.
Byday []PosDayTotal `json:"byday"`
// Which tills contributed, so an outlet with several counters can see them
// apart without a second query.
Byterminal []PosTerminalTotal `json:"byterminal"`
}
type PosPaymentTotal struct {
Paymentmode string `json:"paymentmode"`
Billcount int `json:"billcount"`
Amount float64 `json:"amount"`
}
type PosDayTotal struct {
Businessdate string `json:"businessdate"`
Billcount int `json:"billcount"`
Amount float64 `json:"amount"`
}
type PosTerminalTotal struct {
Terminalid string `json:"terminalid"`
Billcount int `json:"billcount"`
Amount float64 `json:"amount"`
}

12
models/product.go
View File

@@ -77,8 +77,16 @@ type Products struct {
Productcombo int `json:"productcombo" gorm:"default:0"`
Variants int `json:"variants" gorm:"default:0"`
Quantity int `json:"quantity"`
Retailprice float64 `json:"retailprice,omitempty"`
Diffprice float64 `json:"diffprice,omitempty"`
// Price is the EFFECTIVE selling price at the location a query was scoped
// to: productlocations.price when the store has set one, otherwise the
// master Retailprice below. Read-only — it is computed by the query, never
// written through this struct. Location-scoped endpoints must expose it, or
// a price the admin sets per store can never reach the customer app: they
// returned only Retailprice, which the admin catalogue never writes.
// Same meaning as Locationproducts.Price, so both product feeds agree.
Price float64 `json:"price" gorm:"->"`
Retailprice float64 `json:"retailprice,omitempty"`
Diffprice float64 `json:"diffprice,omitempty"`
Diffpercent float64 `json:"diffpercent,omitempty"`
Othercost float64 `json:"othercost,omitempty"`
Approve int `json:"approve"`

135
repositories/orderRepository.go
View File

@@ -1383,6 +1383,110 @@ func (r *orderRepository) reloadOrder(orderHeaderID int) (models.Orders, error)
// use tx again. On success tx is left open and uncommitted, so the caller can
// include its own work — a duplicate-bill guard, an advisory lock — in the same
// transaction as the order that work protects.
// priceOrderLines fills in any line the client sent without a price, using the
// merchant's own catalogue, and brings the header totals in line with the
// result. It mutates data in place and is a no-op for an order that already
// arrived fully priced.
//
// The arithmetic deliberately matches the offline-sales import exactly — gross,
// minus discount, with tax extracted from the resulting landing amount because
// shelf prices here are MRP (tax already inside). One convention for both
// channels, so the same basket rings up the same either way.
func (r *orderRepository) priceOrderLines(tx *gorm.DB, data *models.Orders, defaultLocID int) error {
if len(data.Items) == 0 {
return nil
}
// One catalogue read per outlet, not per line. Items usually share an
// outlet, but a line may name its own.
catalogues := make(map[int]map[int]offlineProduct)
catalogueFor := func(locationID int) (map[int]offlineProduct, error) {
if c, ok := catalogues[locationID]; ok {
return c, nil
}
c, err := loadCatalogueProducts(tx, data.Tenantid, locationID)
if err != nil {
return nil, err
}
catalogues[locationID] = c
return c, nil
}
var lineTotal, taxTotal float64
for i := range data.Items {
item := &data.Items[i]
itemLocID := item.Locationid
if itemLocID == 0 {
itemLocID = defaultLocID
}
if item.Price <= 0 {
catalogue, err := catalogueFor(itemLocID)
if err != nil {
return err
}
// A miss can't normally happen — the stock check above already
// proved the product is stocked here. If it somehow does, leave the
// line as the client sent it rather than refusing the order: a
// pricing lookup is not a reason to block a customer's checkout.
if product, ok := catalogue[item.Productid]; ok {
item.Price = product.Price
if item.Taxpercentage <= 0 {
item.Taxpercentage = product.Taxpercent
}
if item.Productname == "" {
item.Productname = product.Productname
}
}
}
gross := item.Price * item.Orderqty
discount := item.Discountamount
if discount < 0 {
discount = 0
}
if discount > gross {
discount = gross
}
landing := gross - discount
// Only derive what the client didn't state, so a client that does its
// own (possibly promotional) maths keeps its figures.
if item.Productsumprice <= 0 {
item.Productsumprice = gross
}
if item.Landingamount <= 0 {
item.Landingamount = landing
}
if item.Taxamount <= 0 && item.Taxpercentage > 0 {
item.Taxamount = landing - (landing / (1 + item.Taxpercentage/100))
}
lineTotal += item.Landingamount
taxTotal += item.Taxamount
}
// Header totals are only derived when the client left them empty; an order
// that states its own total (delivery charges, promotions applied basket-
// wide) keeps it.
if data.Orderamount <= 0 {
data.Orderamount = float32(lineTotal)
}
if data.Ordervalue <= 0 {
data.Ordervalue = float32(lineTotal)
}
if data.Taxamount <= 0 {
data.Taxamount = float32(taxTotal)
}
if data.Itemcount <= 0 {
data.Itemcount = len(data.Items)
}
return nil
}
func (r *orderRepository) createOrderTx(tx *gorm.DB, data models.Orders) (models.Orders, error) {
locID := data.Locationid
if locID == 0 {
@@ -1424,6 +1528,22 @@ func (r *orderRepository) createOrderTx(tx *gorm.DB, data models.Orders) (models
return models.Orders{}, err
}
// 🛠️ Step 1b: Price the lines the client left unpriced.
//
// Line prices arrive from the client, and a client that sends none books the
// order at zero — which is exactly what happened to every catalogue-imported
// product, whose per-store price was never set: real orders were written
// with price 0 and orderamount 0, so a delivered sale recorded no revenue.
//
// Only lines the client left at or below zero are filled. A line that came
// with a price keeps it, because variants, addons and promotions legitimately
// charge something other than the shelf price and this is not the place to
// second-guess them.
if err := r.priceOrderLines(tx, &data, locID); err != nil {
tx.Rollback()
return models.Orders{}, err
}
// 🛠️ Step 2: Create Order Header
// Claimed inside tx so the row lock on the counter holds until commit:
// concurrent orders queue for it instead of reading the same number, and a
@@ -1632,8 +1752,21 @@ type offlineProduct struct {
// way the line is refused — so a hand-edited productid cannot reach into a
// catalogue the uploader has no claim on.
func (r *orderRepository) loadOfflineProducts(tenantID, locationID int) (map[int]offlineProduct, error) {
return loadCatalogueProducts(r.db, tenantID, locationID)
}
// loadCatalogueProducts is the shared price/tax lookup: the merchant's own
// selling price for every product stocked at one outlet, preferring the
// per-store productlocations.price and falling back to the master
// products.retailprice. Both order paths price from this one query so an online
// order and a counter sale can never disagree about what a product costs.
//
// Takes its handle so a caller inside a transaction reads through that
// transaction — createOrderTx has already locked these product rows, and
// reading around the lock would defeat the point.
func loadCatalogueProducts(db *gorm.DB, tenantID, locationID int) (map[int]offlineProduct, error) {
rows := make([]offlineProduct, 0)
err := r.db.Raw(`
err := db.Raw(`
SELECT a.productid,
COALESCE(a.productname, '') AS productname,
COALESCE(a.productunit, '') AS productunit,

201
repositories/posRepository.go
View File

@@ -37,6 +37,12 @@ type PosRepository interface {
IngestOrders(batch models.PosOrderBatch) (*models.PosAck, error)
IngestCustomers(batch models.PosCustomerBatch) (*models.PosAck, error)
Catalogue(storeID, since string, page, pageSize int) (*models.PosCatalogueResponse, error)
// Reading counter sales back out. Without these a committed bill is
// unreachable from every screen in the product.
Sales(f models.PosSalesFilter) (*models.PosSalesPage, error)
SaleDetail(locationID int, reference string) (*models.PosOrders, error)
SalesSummary(f models.PosSalesFilter) (*models.PosSalesSummary, error)
}
type posRepository struct {
@@ -106,7 +112,7 @@ func (r *posRepository) IngestOrders(batch models.PosOrderBatch) (*models.PosAck
ack.Reject("", "order is missing its id")
continue
}
if reason := r.importPosOrder(ctx, products, batch.Batchid, order); reason != "" {
if reason := r.importPosOrder(ctx, products, batch.Batchid, batch.Terminalid, order); reason != "" {
ack.Reject(order.Id, reason)
continue
}
@@ -126,10 +132,15 @@ func (r *posRepository) IngestOrders(batch models.PosOrderBatch) (*models.PosAck
// in opposite directions and both matter: the bill is its own kind of document
// and deserves its own table, but stock is one number per shelf and must not be
// tracked twice.
// batchTerminal is the terminal the whole batch came from, used when a bill
// does not name one itself. Over MQTT the consumer fills it in from the topic;
// over HTTP the terminal sends it once at the top of the batch rather than
// repeating it on every bill.
func (r *posRepository) importPosOrder(
ctx *offlineLocationContext,
products map[int]offlineProduct,
batchID string,
batchTerminal string,
order models.PosOrder,
) string {
if len(order.Items) == 0 {
@@ -296,7 +307,11 @@ func (r *posRepository) importPosOrder(
Invoicenumber: order.Invoicenumber,
Tenantid: ctx.Tenantid,
Locationid: ctx.Locationid,
Terminalid: order.Terminalid,
// The bill's own terminal wins; the batch's is the fallback. Without
// this the column was empty on every bill that arrived over HTTP —
// the invoice number carried the code and the column did not, so
// per-terminal reconciliation had nothing to group on.
Terminalid: posTerminalFor(order.Terminalid, batchTerminal),
Cashiername: order.Cashier,
Customerid: customerID,
Customermobile: posCustomerMobile(order),
@@ -366,6 +381,18 @@ func posJSON(v any) string {
return string(body)
}
// posTerminalFor picks which terminal code to file a bill under.
//
// Trimmed before the emptiness test: a terminal sending `" "` is saying nothing,
// and treating that as a real code would file bills under a blank that looks
// identical to the missing value this exists to fix.
func posTerminalFor(orderTerminal, batchTerminal string) string {
if t := strings.TrimSpace(orderTerminal); t != "" {
return t
}
return strings.TrimSpace(batchTerminal)
}
func posCustomerName(order models.PosOrder) string {
if order.Customer == nil {
return ""
@@ -385,6 +412,20 @@ func posCustomerMobile(order models.PosOrder) string {
// The terminal sends ISO-8601. A blank one falls back to now; an unparseable
// one is refused, because importing a sale under the wrong date corrupts every
// daily revenue figure that reads it.
// parsePosSaleDate reads the moment a bill was rung.
//
// The order of these layouts is load-bearing, and the two zoned ones must stay
// first. A terminal that sends its offset — `2026-08-05T00:30:00+05:30` — gets
// both readings right: the instant is correct, and Format("2006-01-02") still
// yields the till's own trading day rather than UTC's.
//
// The two bare layouts exist for terminals built before the offset was added,
// which are still in the field. `time.Parse` fills an absent zone with UTC, so
// those bills record an instant wrong by the offset — a Coimbatore wall clock
// read as though it were London. That is not recoverable here: nothing in the
// payload says which zone it came from. Their business date is still right,
// which is why the daily figures held up while billedat did not, and why these
// are tolerated rather than refused.
func parsePosSaleDate(raw string) (time.Time, error) {
raw = strings.TrimSpace(raw)
if raw == "" {
@@ -488,12 +529,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 {
@@ -507,6 +603,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
@@ -521,8 +622,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,
@@ -542,12 +660,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
}
@@ -560,14 +681,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
@@ -606,15 +725,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
}

158
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,156 @@ 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)
}
}
}
// The bug this covers reached production and stayed invisible for a day.
//
// The MQTT consumer backfills a missing terminal code from the topic, but it
// wrote it onto the *batch* while the row was built from the *order*, so the
// two never met. Bills arriving over HTTP had no topic to fall back on at all.
// The result: 16 of 17 live bills carried an empty terminalid while their own
// invoice numbers read INV-2608-T5EDD-000NN, and `byterminal` on the sales
// summary grouped almost everything under "".
func TestABillTakesItsTerminalFromTheBatchWhenItNamesNone(t *testing.T) {
cases := []struct {
name string
orderTerminal string
batchTerminal string
want string
}{
{"bill names its own", "T5EDD", "TOTHER", "T5EDD"},
{"bill is silent, batch knows", "", "T5EDD", "T5EDD"},
{"neither knows", "", "", ""},
// Whitespace is not a terminal code. Treating it as one would file
// bills under a blank that reads identically to the missing value
// this fallback exists to prevent.
{"bill sends whitespace", " ", "T5EDD", "T5EDD"},
{"batch sends whitespace", "", " ", ""},
{"codes are trimmed", " T5EDD ", "", "T5EDD"},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
if got := posTerminalFor(c.orderTerminal, c.batchTerminal); got != c.want {
t.Errorf("posTerminalFor(%q, %q) = %q, want %q",
c.orderTerminal, c.batchTerminal, got, c.want)
}
})
}
}
// billedat and businessdate are derived from the same parsed value and pull in
// opposite directions, so they are tested together.
//
// Live bill INV-2608-T5EDD-00116 carried billedat 2026-08-05T12:49:28Z beside
// receivedat 2026-08-05T07:19:28Z — the sale appearing to happen five and a
// half hours after it was received. The till was sending a naive local
// timestamp and time.Parse fills that silence with UTC, so a Coimbatore wall
// clock was recorded as though read in London.
//
// The daily figures survived it by luck: businessdate comes off the wall clock
// either way, and the wall clock was always the till's own. Anything comparing
// billedat against real time did not.
func TestASaleDateKeepsBothTheInstantAndTheTradingDay(t *testing.T) {
// Coimbatore, late enough that UTC has not yet rolled into the same day.
const ist = "2026-08-05T00:30:00+05:30"
at, err := parsePosSaleDate(ist)
if err != nil {
t.Fatalf("parsePosSaleDate(%q) errored: %v", ist, err)
}
// The instant. 00:30 IST is 19:00 UTC the previous evening.
wantInstant := time.Date(2026, 8, 4, 19, 0, 0, 0, time.UTC)
if !at.UTC().Equal(wantInstant) {
t.Errorf("instant = %v, want %v", at.UTC(), wantInstant)
}
// The trading day. This is the one that must NOT follow UTC — the shop rang
// this sale on the 5th and its takings belong to the 5th. Deriving the
// business date from UTC would file it under the 4th and leave two days
// wrong: one short, one over.
if got := at.Format("2006-01-02"); got != "2026-08-05" {
t.Errorf("businessdate = %s, want 2026-08-05 — the till's own day", got)
}
}
// Terminals built before the offset was added send a bare local timestamp, and
// they are still in the field. Parsing must not start refusing them.
//
// The instant such a bill records is wrong by the offset and cannot be
// recovered — there is nothing in the payload that says which zone it was read
// in. Its business date is still right, which is why the daily figures held up,
// and why this stays a tolerated legacy rather than a rejection.
func TestANaiveSaleDateIsStillAccepted(t *testing.T) {
at, err := parsePosSaleDate("2026-08-05T12:49:28.245")
if err != nil {
t.Fatalf("a pre-offset terminal must not be refused: %v", err)
}
if got := at.Format("2006-01-02"); got != "2026-08-05" {
t.Errorf("businessdate = %s, want 2026-08-05", got)
}
}

223
repositories/posSalesRepository.go Normal file
View File

@@ -0,0 +1,223 @@
package repositories
import (
"fmt"
"strings"
"nearle/models"
)
// Reading counter sales back out.
//
// The ingest side of this package only ever writes. Without these, a bill that
// reached pos_orders was invisible to every screen in the product — the data
// was safe and unreachable, which is its own kind of lost.
//
// Every query is scoped to one locationid. That is the authorisation boundary:
// a caller who omits it gets an error rather than a page through somebody
// else's takings.
// posSalesWhere builds the shared filter, so the list, the detail and the
// summary can never disagree about what "this outlet in this range" means.
func posSalesWhere(f models.PosSalesFilter) (string, []interface{}) {
where := "locationid = ?"
params := []interface{}{f.Locationid}
// Matched on businessdate — the day the sale was rung, not the day it
// reached us. A till that was offline overnight uploads yesterday's bills
// this morning and they belong to yesterday.
if f.Fromdate != "" && f.Todate != "" {
where += " AND businessdate BETWEEN ? AND ?"
params = append(params, f.Fromdate, f.Todate)
} else if f.Fromdate != "" {
where += " AND businessdate >= ?"
params = append(params, f.Fromdate)
} else if f.Todate != "" {
where += " AND businessdate <= ?"
params = append(params, f.Todate)
}
if t := strings.TrimSpace(f.Terminalid); t != "" {
where += " AND terminalid = ?"
params = append(params, t)
}
if c := strings.TrimSpace(f.Cashiername); c != "" {
where += " AND cashiername = ?"
params = append(params, c)
}
if p := strings.TrimSpace(f.Paymentmode); p != "" {
where += " AND LOWER(paymentmode) = ?"
params = append(params, strings.ToLower(p))
}
return where, params
}
// Sales returns a page of bills, newest first, with the total count.
//
// Line items are deliberately not included: a page of fifty bills would drag
// several hundred rows behind it, and a list screen shows none of them. Use
// SaleDetail for one bill.
func (r *posRepository) Sales(f models.PosSalesFilter) (*models.PosSalesPage, error) {
if f.Locationid <= 0 {
return nil, fmt.Errorf("locationid is required")
}
if f.Pagesize <= 0 || f.Pagesize > 500 {
f.Pagesize = 50
}
if f.Pageno < 0 {
f.Pageno = 0
}
where, params := posSalesWhere(f)
var total int64
if err := r.db.Raw(
fmt.Sprintf(`SELECT COUNT(*) FROM pos_orders WHERE %s`, where),
params...,
).Scan(&total).Error; err != nil {
return nil, err
}
bills := make([]models.PosOrders, 0)
// Ordered by billedat rather than by id: a batch uploaded after an outage
// arrives out of order, and a list sorted by arrival would interleave
// yesterday's bills through today's.
query := fmt.Sprintf(
`SELECT * FROM pos_orders WHERE %s
ORDER BY billedat DESC, posorderid DESC
LIMIT ? OFFSET ?`, where)
if err := r.db.Raw(query,
append(params, f.Pagesize, f.Pageno*f.Pagesize)...,
).Scan(&bills).Error; err != nil {
return nil, err
}
return &models.PosSalesPage{
Total: total,
Pageno: f.Pageno,
Pagesize: f.Pagesize,
Bills: bills,
}, nil
}
// SaleDetail returns one bill with its lines.
//
// Accepts either the terminal's own order UUID or this backend's posorderid,
// because a support call starts from whichever the caller happens to be looking
// at — a receipt carries the invoice number, a log carries the UUID.
func (r *posRepository) SaleDetail(locationID int, reference string) (*models.PosOrders, error) {
if locationID <= 0 {
return nil, fmt.Errorf("locationid is required")
}
reference = strings.TrimSpace(reference)
if reference == "" {
return nil, fmt.Errorf("an order id, invoice number or posorderid is required")
}
var bill models.PosOrders
err := r.db.Raw(`
SELECT * FROM pos_orders
WHERE locationid = ?
AND (terminalorderid = ? OR invoicenumber = ?
OR CAST(posorderid AS TEXT) = ?)
LIMIT 1`,
locationID, reference, reference, reference,
).Scan(&bill).Error
if err != nil {
return nil, err
}
if bill.Posorderid == 0 {
return nil, nil
}
items := make([]models.PosOrderItems, 0)
if err := r.db.Raw(
`SELECT * FROM pos_order_items WHERE posorderid = ? ORDER BY posorderitemid`,
bill.Posorderid,
).Scan(&items).Error; err != nil {
return nil, err
}
bill.Items = items
return &bill, nil
}
// SalesSummary totals a range, broken out the three ways somebody actually
// asks for: by tender, by day, and by till.
func (r *posRepository) SalesSummary(f models.PosSalesFilter) (*models.PosSalesSummary, error) {
if f.Locationid <= 0 {
return nil, fmt.Errorf("locationid is required")
}
where, params := posSalesWhere(f)
summary := &models.PosSalesSummary{
Locationid: f.Locationid,
Fromdate: f.Fromdate,
Todate: f.Todate,
Bypaymentmode: make([]models.PosPaymentTotal, 0),
Byday: make([]models.PosDayTotal, 0),
Byterminal: make([]models.PosTerminalTotal, 0),
}
var head struct {
Billcount int
Itemcount int
Grosssales float64
Taxcollected float64
Discount float64
Roundoff float64
}
if err := r.db.Raw(fmt.Sprintf(`
SELECT COUNT(*) AS billcount,
COALESCE(SUM(itemcount), 0) AS itemcount,
COALESCE(SUM(total), 0) AS grosssales,
COALESCE(SUM(taxamount), 0) AS taxcollected,
COALESCE(SUM(discount), 0) AS discount,
COALESCE(SUM(roundoff), 0) AS roundoff
FROM pos_orders WHERE %s`, where), params...).Scan(&head).Error; err != nil {
return nil, err
}
summary.Billcount = head.Billcount
summary.Itemcount = head.Itemcount
summary.Grosssales = head.Grosssales
summary.Taxcollected = head.Taxcollected
summary.Discount = head.Discount
summary.Roundoff = head.Roundoff
if head.Billcount > 0 {
summary.Averagebill = head.Grosssales / float64(head.Billcount)
}
if err := r.db.Raw(fmt.Sprintf(`
SELECT COALESCE(paymentmode,'') AS paymentmode,
COUNT(*) AS billcount, COALESCE(SUM(total),0) AS amount
FROM pos_orders WHERE %s
GROUP BY paymentmode ORDER BY amount DESC`, where),
params...).Scan(&summary.Bypaymentmode).Error; err != nil {
return nil, err
}
if err := r.db.Raw(fmt.Sprintf(`
SELECT businessdate, COUNT(*) AS billcount,
COALESCE(SUM(total),0) AS amount
FROM pos_orders WHERE %s
GROUP BY businessdate ORDER BY businessdate`, where),
params...).Scan(&summary.Byday).Error; err != nil {
return nil, err
}
if err := r.db.Raw(fmt.Sprintf(`
SELECT COALESCE(terminalid,'') AS terminalid, COUNT(*) AS billcount,
COALESCE(SUM(total),0) AS amount
FROM pos_orders WHERE %s
GROUP BY terminalid ORDER BY amount DESC`, where),
params...).Scan(&summary.Byterminal).Error; err != nil {
return nil, err
}
return summary, nil
}

29
repositories/productRepository.go
View File

@@ -465,7 +465,13 @@ func (r *productRepository) GetLocationProducts(tenantID, locationID, subcategor
// into Locationproducts.Quantity, which is what made the console's stock
// column look frozen after an order despite CreateOrder recording the
// "out" ledger entry correctly.
query := `SELECT a.*, b.productlocationid, b.status, b.price,
// COALESCE so `price` means the same thing here as in GetProducts: the
// effective selling price at this outlet, falling back to the master
// retailprice when the store hasn't set its own. Returning a bare b.price
// reported 0 for any product priced only at tenant level, which the store
// catalogue then rendered as "—".
query := `SELECT a.*, b.productlocationid, b.status,
COALESCE(NULLIF(b.price, 0), a.retailprice, 0) AS price,
COALESCE(SUM(CASE WHEN UPPER(c.stocktype) = 'IN' THEN c.quantity ELSE 0 END), 0) AS total_in,
COALESCE(SUM(CASE WHEN UPPER(c.stocktype) = 'OUT' THEN c.quantity ELSE 0 END), 0) AS total_out,
COALESCE(SUM(CASE WHEN UPPER(c.stocktype) = 'IN' THEN c.quantity ELSE 0 END) -
@@ -842,7 +848,7 @@ func (r *productRepository) GetProducts(params models.ProductFilter) ([]models.P
var products []models.Products
q := r.db.Table("products a").
Joins("LEFT JOIN productlocations pl ON pl.productid = a.productid").
Joins("LEFT JOIN productlocations pl ON pl.productid = a.productid AND pl.tenantid = a.tenantid").
Joins("LEFT JOIN productdiscounts pd ON pd.productid = a.productid").
Joins("LEFT JOIN productcategories c ON a.categoryid = c.categoryid").
Where("a.categoryid = ?", params.CategoryID)
@@ -875,9 +881,26 @@ func (r *productRepository) GetProducts(params models.ProductFilter) ([]models.P
// never-decremented products.quantity column — same fix as
// GetLocationProducts/GetProductByVariant, otherwise this endpoint would
// keep showing stock that never reduces after an order.
// price is the effective selling price at params.LocationID: the store's own
// productlocations.price, falling back to the master products.retailprice
// when that outlet hasn't set one. It has to be here — this endpoint feeds
// the customer app's browse-by-subcategory view, and `a.*` only carries
// retailprice, which the admin catalogue never writes. So a price the admin
// set per store could never reach the app; every product priced as 0.
//
// Deliberately a correlated subquery rather than a read off the joined `pl`:
// that join isn't outlet-scoped unless params.LocationID is set, so reading
// pl.price directly would pick an arbitrary branch's price (and multiply the
// rows) whenever the caller didn't scope to one. Same shape as the
// productstock subqueries below, for the same reason.
err := q.Select(`
a.*,
COALESCE(pd.discountvalue, 0) AS discountvalue,
COALESCE(NULLIF((
SELECT pl2.price FROM productlocations pl2
WHERE pl2.productid = a.productid AND pl2.tenantid = a.tenantid AND pl2.locationid = ?
LIMIT 1
), 0), a.retailprice, 0) AS price,
COALESCE((
SELECT SUM(CASE WHEN LOWER(ps.stocktype) = 'in' THEN ps.quantity ELSE 0 END) -
SUM(CASE WHEN LOWER(ps.stocktype) = 'out' THEN ps.quantity ELSE 0 END)
@@ -890,7 +913,7 @@ func (r *productRepository) GetProducts(params models.ProductFilter) ([]models.P
FROM productstocks ps
WHERE ps.productid = a.productid AND ps.tenantid = a.tenantid AND ps.locationid = ?
), 0) AS quantity
`, params.LocationID, params.LocationID).Find(&products).Error
`, params.LocationID, params.LocationID, params.LocationID).Find(&products).Error
return products, err
}

11
routes/posroutes.go
View File

@@ -23,6 +23,17 @@ func RegisterPosRoutes(api fiber.Router, f *facade.Facade) {
pos.Post("/customers", f.PosController.IngestCustomers)
pos.Get("/catalogue", f.PosController.Catalogue)
// The 30-second heartbeat, for tills on the HTTP route. The broker carries
// the same payload for tills on MQTT; both land in the same Redis record,
// so the fleet board cannot tell them apart and does not need to.
pos.Post("/health", f.PosController.IngestHealth)
// Counter sales, read back out. The ingest above only ever writes; without
// these a committed bill is unreachable from every screen in the product.
pos.Get("/sales", f.PosController.GetSales)
pos.Get("/sales/detail", f.PosController.GetSaleDetail)
pos.Get("/sales/summary", f.PosController.GetSalesSummary)
// Terminal presence, read from Redis. What the rider app's POS board and a
// support call both hit — the tills themselves publish health over the
// broker rather than posting it here.

28
scratch/dbinspect/cleanup.go
View File

@@ -15,7 +15,8 @@ import (
// The two test bills are named explicitly rather than deleted by date or by
// "everything in pos_orders" — a table that will hold real takings tomorrow is
// not one to run an unbounded DELETE against.
const probeMobile = "9840012345"
// Every mobile the probes registered, so a cleanup run leaves nothing behind.
var probeMobiles = []string{"9840012345", "9840099999", "9840077777"}
var testOrderIDs = []string{
"11111111-2222-4333-8444-555555555555", // the HTTP probe
@@ -70,16 +71,21 @@ func cleanup(db *gorm.DB) {
// The shopper the probes created. Removed only when nothing references it —
// a customer row attached to a real order is not test data any more.
var referenced int
db.Raw(`SELECT COUNT(*) FROM orders WHERE customerid =
(SELECT MIN(customerid) FROM customers WHERE contactno = ?)`,
probeMobile).Scan(&referenced)
if referenced > 0 {
fmt.Printf(" customer %s left in place — %d order(s) reference it\n",
probeMobile, referenced)
} else {
res := db.Exec(`DELETE FROM customers WHERE contactno = ?`, probeMobile)
fmt.Printf(" removed %d probe customer row(s)\n", res.RowsAffected)
for _, mobile := range probeMobiles {
// A customer row attached to a real order is not test data any more.
var referenced int
db.Raw(`SELECT COUNT(*) FROM orders WHERE customerid IN
(SELECT customerid FROM customers WHERE contactno = ?)`,
mobile).Scan(&referenced)
if referenced > 0 {
fmt.Printf(" customer %s left in place — %d order(s) reference it\n",
mobile, referenced)
continue
}
res := db.Exec(`DELETE FROM customers WHERE contactno = ?`, mobile)
if res.RowsAffected > 0 {
fmt.Printf(" removed probe customer %s\n", mobile)
}
}
var balance float64

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

180
scratch/gstrates/main.go Normal file
View File

@@ -0,0 +1,180 @@
// Set GST rates on the POS catalogue products.
//
// The four packaged lines at 1185 sit at taxpercent 0 and are being billed with
// no GST at all — a live compliance problem rather than a cosmetic one. Those
// are written.
//
// The produce at 1135 is NOT all zero, which is what this was first written
// believing. It holds 8, 12 and 18, and under Indian GST fresh unbranded fruit
// and chilled fish are nil-rated — so several look like overcharging. Every
// correction there is a *reduction* of a live rate, which is a decision for
// whoever signs the returns. Reported as REVIEW and left untouched.
//
// go run ./scratch/gstrates plan
// go run ./scratch/gstrates apply
package main
import (
"fmt"
"log"
"os"
"github.com/joho/godotenv"
"gorm.io/driver/postgres"
"gorm.io/gorm"
"gorm.io/gorm/logger"
)
// Indian GST on food, as it applies to these lines.
//
// Fresh, unbranded and unpackaged produce is nil-rated, which is why the fruit
// stays at 0 rather than being "not set yet". Packaged branded snacks are 12%.
// Breakfast cereal is 18%.
//
// Fish is the one worth stating: fresh or chilled is nil-rated, and only
// frozen/branded/packaged attracts 5%. Left at 0 on the reading that a counter
// selling loose Mysore bananas is selling fresh fish, not frozen packs.
type rate struct {
productID int
name string
percent float64
why string
// apply gates the write. Only rows that are unambiguously *unset* are
// written; anything already carrying a rate is reported and left alone.
//
// The fresh produce at 1135 is the reason for this flag. Those rows are not
// blank — they hold 8, 12 and 18 — and under Indian GST fresh unbranded
// fruit and chilled fish are nil-rated, so several look like overcharging.
// But *lowering* a live tax rate is a compliance decision belonging to
// whoever signs the returns, not a bug to be quietly corrected by a script,
// and someone is actively working on pricing in this repo. Reported, not
// touched.
apply bool
}
var rates = []rate{
// 1135 — fresh produce, nil-rated under Indian GST.
//
// These were held back at first because every one is a *reduction* of a
// live rate, which is a compliance decision rather than a bug fix. Released
// on the owner's explicit instruction after that was put to them.
//
// Two readings are assumed and should be checked against what the counter
// actually sells: Maceral and Tuna are taken as fresh or chilled, which is
// nil-rated — frozen, branded or packaged fish is 5%. Hatsun curd is taken
// as plain curd, which is nil-rated — flavoured yoghurt is 5%.
{6988, "Mysore Banana", 0, "fresh fruit — nil-rated, currently 8%", true},
{6989, "Jammu Apple", 0, "fresh fruit — nil-rated, currently 18%", true},
{6990, "Small orange", 0, "fresh fruit — nil-rated, currently 18%", true},
{6991, "Red Guava", 0, "fresh fruit — nil-rated, currently 18%", true},
{6992, "Pomegrante", 0, "fresh fruit — nil-rated, currently 12%", true},
{6993, "Salem Mango", 0, "fresh fruit — nil-rated", true},
{6994, "Pineapple", 0, "fresh fruit — nil-rated", true},
{6995, "Strawberries", 0, "fresh fruit — nil-rated, currently 18%", true},
{6996, "Maceral", 0, "fresh fish nil-rated; 5% only if frozen/packaged", true},
{6997, "Tuna", 0, "fresh fish nil-rated; 5% only if frozen/packaged", true},
{6998, "Hatsun curd", 0, "curd nil-rated; flavoured yoghurt would be 5%", true},
{7014, "Apple", 0, "fresh fruit — nil-rated", true},
// 1185 — genuinely unset, and being billed with no GST at all today. This
// is the half that is unambiguous: every one is an increase from zero, so
// nothing is being under-collected on the strength of a script's opinion.
{7074, "Amla Dabur Oral Care Chewing Gum 10g", 18, "chewing gum, 18%", true},
{7075, "Cheetos Chips 100g", 12, "packaged extruded snack, 12%", true},
{7076, "Cheerios Breakfast Cereal 100g", 18, "packaged cereal, 18%", true},
{7077, "Hot Heads 30g", 12, "packaged snack, 12%", true},
}
func main() {
mode := "plan"
if len(os.Args) > 1 {
mode = os.Args[1]
}
_ = godotenv.Load()
dsn := fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable",
os.Getenv("DB_HOST"), os.Getenv("DB_PORT"), os.Getenv("DB_USER"),
os.Getenv("DB_PASSWORD"), os.Getenv("DB_NAME"))
db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{
Logger: logger.Default.LogMode(logger.Silent),
})
if err != nil {
log.Fatal(err)
}
write := mode == "apply"
fmt.Printf("%-6s %-38s %6s -> %6s %s\n", "id", "product", "now", "new", "why")
fmt.Println("-------------------------------------------------------------------------------------------")
undo := []string{}
changes := 0
review := 0
for _, r := range rates {
var current struct {
Taxpercent float64
Found bool
}
if err := db.Raw(`SELECT COALESCE(taxpercent, 0) AS taxpercent, true AS found
FROM products WHERE productid = ? LIMIT 1`,
r.productID).Scan(&current).Error; err != nil {
log.Fatalf("reading %d: %v", r.productID, err)
}
if !current.Found {
fmt.Printf("%-6d %-38s NO products ROW - skipped\n", r.productID, r.name)
continue
}
if current.Taxpercent == r.percent {
fmt.Printf("%-6d %-38s %6.0f unchanged %s\n",
r.productID, r.name, current.Taxpercent, r.why)
continue
}
if !r.apply {
fmt.Printf("%-6d %-38s %6.0f REVIEW %-3.0f %s\n",
r.productID, r.name, current.Taxpercent, r.percent, r.why)
review++
continue
}
fmt.Printf("%-6d %-38s %6.0f -> %6.0f %s\n",
r.productID, r.name, current.Taxpercent, r.percent, r.why)
undo = append(undo, fmt.Sprintf(
"UPDATE products SET taxpercent = %.0f WHERE productid = %d;",
current.Taxpercent, r.productID))
changes++
if write {
// updated is bumped so the catalogue delta carries the new rate to
// terminals holding a revision, rather than waiting for a full pull.
if err := db.Exec(`UPDATE products SET taxpercent = ?, updated = NOW()
WHERE productid = ?`, r.percent, r.productID).Error; err != nil {
log.Fatalf("writing %d: %v", r.productID, err)
}
}
}
fmt.Println("-------------------------------------------------------------------------------------------")
if write {
fmt.Printf("APPLIED %d rate(s).\n", changes)
} else {
fmt.Printf("%d rate(s) would change. Nothing written — run `apply` to commit.\n", changes)
}
if review > 0 {
fmt.Printf("%d row(s) flagged REVIEW and deliberately not written — each is a\n"+
"reduction of a live tax rate and needs a decision, not a script.\n", review)
}
fmt.Println()
if len(undo) > 0 {
fmt.Println("-- undo:")
for _, u := range undo {
fmt.Println(u)
}
}
}

141
scratch/healthproof/main.go Normal file
View File

@@ -0,0 +1,141 @@
// Proof that the 30-second health heartbeat works end to end, over MQTT.
//
// Publishes a heartbeat to the live broker as a throwaway terminal, then polls
// the public API until it shows online — and keeps polling past the TTL so the
// automatic expiry is visible too. Nothing is written to Postgres; presence
// lives in Redis under a TTL and cleans itself up.
//
// REQUIRES pos_terminal CREDENTIALS. The MQTT_USER in .env is pos_ingest, which
// the broker ACL deliberately denies publish on the health topic — it may only
// write acks and the catalogue. Running this with the ingest account connects
// fine and then silently drops every publish, because Mosquitto answers an
// ACL-denied QoS 1 publish with a PUBACK and discards it. That looks exactly
// like a broken consumer and cost an hour of misdiagnosis; set MQTT_USER and
// MQTT_PASSWORD to the terminal account before believing a negative result.
//
// The HTTP path needs none of this — see POST /pos/health, which is what the
// fix on v1.3.96 added and how the endpoint was actually verified.
//
// go run ./scratch/healthproof
package main
import (
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"os"
"time"
mqtt "github.com/eclipse/paho.mqtt.golang"
"github.com/joho/godotenv"
)
const (
locationID = "1135"
terminalID = "TPROOF" // throwaway; disappears on its own when the TTL lapses
apiBase = "https://fiesta.nearle.app/live/api/v1/pos"
)
func main() {
_ = godotenv.Load()
brokerURL := os.Getenv("MQTT_URL")
if brokerURL == "" {
log.Fatal("MQTT_URL not set")
}
opts := mqtt.NewClientOptions().
AddBroker(brokerURL).
SetClientID("healthproof-" + terminalID).
SetUsername(os.Getenv("MQTT_USER")).
SetPassword(os.Getenv("MQTT_PASSWORD")).
SetConnectTimeout(10 * time.Second)
client := mqtt.NewClient(opts)
if t := client.Connect(); t.Wait() && t.Error() != nil {
log.Fatal("connect: ", t.Error())
}
defer client.Disconnect(250)
fmt.Printf("connected to %s as %s\n\n", brokerURL, terminalID)
fmt.Println("BEFORE — has this terminal ever been seen?")
show()
// Exactly what the till sends every 30 seconds.
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": 17, "today_amount": 2510.0,
"printer_reachable": true,
"reported_at": time.Now().UTC().Format(time.RFC3339),
})
topic := fmt.Sprintf("nearle/pos/%s/%s/health", locationID, terminalID)
if t := client.Publish(topic, 1, false, health); t.Wait() && t.Error() != nil {
log.Fatal("publish: ", t.Error())
}
fmt.Printf("\npublished one heartbeat to %s\n", topic)
time.Sleep(2 * time.Second)
fmt.Println("\nAFTER one heartbeat:")
show()
// The TTL is 90s and a real till refreshes every 30s, so it never lapses
// while the till is alive. Stopping here is what a till being switched off
// looks like.
fmt.Println("\nnow going quiet, as a till that was switched off would.")
fmt.Println("presence TTL is 90s, so it should drop off on its own:")
for _, wait := range []int{30, 30, 35} {
time.Sleep(time.Duration(wait) * time.Second)
fmt.Printf("\n+%ds since the last heartbeat:\n", wait)
show()
}
}
func show() {
resp, err := http.Get(fmt.Sprintf("%s/health/location?location_id=%s", apiBase, locationID))
if err != nil {
fmt.Println(" API unreachable:", err)
return
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
var out struct {
Details struct {
Online int `json:"online"`
Total int `json:"total"`
Terminals []struct {
Terminalid string `json:"terminal_id"`
Status string `json:"status"`
Reason string `json:"reason"`
Todaybills int `json:"today_bills"`
Todayamount float64 `json:"today_amount"`
Reportedat string `json:"reported_at"`
} `json:"terminals"`
} `json:"details"`
}
if err := json.Unmarshal(body, &out); err != nil {
fmt.Println(" unparseable:", string(body)[:200])
return
}
fmt.Printf(" online %d of %d\n", out.Details.Online, out.Details.Total)
for _, t := range out.Details.Terminals {
mark := " "
if t.Terminalid == terminalID {
mark = ">"
}
extra := t.Reason
if t.Status == "online" {
extra = fmt.Sprintf("today %d bills / Rs %.0f, reported %s",
t.Todaybills, t.Todayamount, t.Reportedat)
}
fmt.Printf(" %s %-8s %-8s %s\n", mark, t.Terminalid, t.Status, extra)
}
}

28
scratch/mqttpub/main.go
View File

@@ -118,5 +118,31 @@ func main() {
log.Fatal("publish health:", t.Error())
}
fmt.Println("\npublished a heartbeat to", healthTopic)
time.Sleep(2 * time.Second)
// 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)
}
}

194
scratch/seedprices/main.go Normal file
View File

@@ -0,0 +1,194 @@
// Seed retail prices so the POS has something sellable.
//
// These are plausible Coimbatore figures, not authoritative ones. They exist so
// the terminal can ring a real bill; the owner corrects them afterwards.
//
// go run ./scratch/seedprices plan # show every change and the undo SQL
// go run ./scratch/seedprices apply # write them
// go run ./scratch/seedprices verify # read back what the catalogue now serves
package main
import (
"fmt"
"log"
"os"
"github.com/joho/godotenv"
"gorm.io/driver/postgres"
"gorm.io/gorm"
"gorm.io/gorm/logger"
)
type priced struct {
locationID int
productID int
name string
unit string
price float64
note string
}
// Prices are per the product's own unit — per kilogram where the unit is
// kilogram, per pack where it is piece. Getting that backwards is the easiest
// way to make a till look broken, so the unit is carried through and printed.
var seed = []priced{
// 1135 — fresh produce
{1135, 6988, "Mysore Banana", "kilogram", 60, ""},
{1135, 6989, "Jammu Apple", "piece", 30, "per fruit, not per kg"},
{1135, 6990, "Small orange", "kilogram", 90, ""},
{1135, 6991, "Red Guava", "kilogram", 80, ""},
{1135, 6992, "Pomegrante", "kilogram", 180, ""},
{1135, 6993, "Salem Mango", "kilogram", 90, "seasonal, swings 80-120"},
{1135, 6994, "Pineapple", "kilogram", 60, ""},
{1135, 6995, "Strawberries", "piece", 150, "priced as a punnet"},
{1135, 6996, "Maceral", "kilogram", 220, "READ AS MACKEREL - correct if wrong"},
{1135, 6997, "Tuna", "kilogram", 280, ""},
{1135, 6998, "Hatsun curd", "piece", 30, "500g pouch"},
{1135, 7014, "Apple", "kilogram", 200, ""},
// 1185 — packaged
{1185, 7074, "Amla Dabur Oral Care Chewing Gum 10g", "piece", 10, ""},
{1185, 7075, "Cheetos Chips 100g", "piece", 40, ""},
{1185, 7076, "Cheerios Breakfast Cereal 100g", "piece", 120, ""},
// 7077 Hot Heads is already at 50 — someone set it deliberately, leave it.
}
func main() {
mode := "plan"
if len(os.Args) > 1 {
mode = os.Args[1]
}
_ = godotenv.Load()
dsn := fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable",
os.Getenv("DB_HOST"), os.Getenv("DB_PORT"), os.Getenv("DB_USER"),
os.Getenv("DB_PASSWORD"), os.Getenv("DB_NAME"))
db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{
Logger: logger.Default.LogMode(logger.Silent),
})
if err != nil {
log.Fatal(err)
}
switch mode {
case "plan":
plan(db, false)
case "apply":
plan(db, true)
case "verify":
verify(db)
default:
log.Fatalf("unknown mode %q — use plan, apply or verify", mode)
}
}
func plan(db *gorm.DB, write bool) {
fmt.Printf("%-6s %-38s %-9s %8s -> %8s\n", "id", "product", "unit", "now", "new")
fmt.Println("--------------------------------------------------------------------------------")
undo := []string{}
changes := 0
for _, p := range seed {
var current struct {
Price float64
Tenantid int
Found bool
}
row := db.Raw(`SELECT COALESCE(price, 0) AS price, tenantid, true AS found
FROM productlocations
WHERE productid = ? AND locationid = ?
LIMIT 1`, p.productID, p.locationID).Scan(&current)
if row.Error != nil {
log.Fatalf("reading %d: %v", p.productID, row.Error)
}
if !current.Found {
fmt.Printf("%-6d %-38s NO productlocations ROW - skipped\n", p.productID, p.name)
continue
}
// Never overwrite a price a human already set. A seed value is a
// placeholder; a real one is a decision, and losing it silently would
// be worse than leaving a gap.
if current.Price > 0 {
fmt.Printf("%-6d %-38s %-9s %8.2f already priced, left alone\n",
p.productID, p.name, p.unit, current.Price)
continue
}
note := ""
if p.note != "" {
note = " <- " + p.note
}
fmt.Printf("%-6d %-38s %-9s %8.2f -> %8.2f%s\n",
p.productID, p.name, p.unit, current.Price, p.price, note)
undo = append(undo, fmt.Sprintf(
"UPDATE productlocations SET price = %.2f WHERE productid = %d AND locationid = %d;",
current.Price, p.productID, p.locationID))
changes++
if write {
// updated is bumped so the catalogue delta carries the new price to
// terminals that already hold a revision, rather than waiting for
// someone to force a full pull.
err := db.Exec(`UPDATE productlocations
SET price = ?, updated = NOW()
WHERE productid = ? AND locationid = ?`,
p.price, p.productID, p.locationID).Error
if err != nil {
log.Fatalf("writing %d: %v", p.productID, err)
}
}
}
fmt.Println("--------------------------------------------------------------------------------")
if write {
fmt.Printf("APPLIED %d price(s).\n\n", changes)
} else {
fmt.Printf("%d price(s) would change. Nothing written — run `apply` to commit.\n\n", changes)
}
fmt.Println("-- undo, if you want the zeros back:")
for _, u := range undo {
fmt.Println(u)
}
}
func verify(db *gorm.DB) {
type row struct {
Locationid int
Productid int
Productname string
Price float64
Taxpercent float64
}
var rows []row
db.Raw(`SELECT b.locationid, a.productid, a.productname,
COALESCE(b.price, 0) AS price, COALESCE(a.taxpercent, 0) AS taxpercent
FROM products a
INNER JOIN productlocations b ON a.productid = b.productid AND a.tenantid = b.tenantid
WHERE b.locationid IN (1135, 1185) AND a.productid > 0
ORDER BY b.locationid, a.productid`).Scan(&rows)
sellable := 0
for _, r := range rows {
flag := ""
if r.Price > 0 {
sellable++
} else {
flag = " <- still zero, not sellable"
}
fmt.Printf("loc %d %-6d %-38s %8.2f tax=%.0f%s\n",
r.Locationid, r.Productid, r.Productname[:min(38, len(r.Productname))], r.Price, r.Taxpercent, flag)
}
fmt.Printf("\n%d of %d rows are sellable.\n", sellable, len(rows))
}
func min(a, b int) int {
if a < b {
return a
}
return b
}

90
scratch/termfixcleanup/main.go Normal file
View File

@@ -0,0 +1,90 @@
// Removes the single bill posted to prove the terminalid fix on v1.3.96.
//
// Named by its own terminalorderid rather than by date or by "the newest row" —
// pos_orders holds real takings, and is not a table to run an unbounded DELETE
// against. Stock is returned before the bill is deleted, so the ledger is never
// left short with nothing remaining to explain why.
//
// go run ./scratch/termfixcleanup
package main
import (
"fmt"
"log"
"os"
"github.com/joho/godotenv"
"gorm.io/driver/postgres"
"gorm.io/gorm"
"gorm.io/gorm/logger"
)
const (
tenantID = 1087
locationID = 1135
testOrder = "a1b2c3d4-0000-4000-8000-termfix00001"
)
func main() {
_ = godotenv.Load()
dsn := fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable",
os.Getenv("DB_HOST"), os.Getenv("DB_PORT"), os.Getenv("DB_USER"),
os.Getenv("DB_PASSWORD"), os.Getenv("DB_NAME"))
db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{
Logger: logger.Default.LogMode(logger.Silent),
})
if err != nil {
log.Fatal(err)
}
var billIDs []int
db.Raw(`SELECT posorderid FROM pos_orders WHERE terminalorderid = ?`,
testOrder).Scan(&billIDs)
if len(billIDs) == 0 {
fmt.Println("no test bill found — nothing to undo")
return
}
fmt.Printf("found test bill(s): %v\n", billIDs)
var consumed []struct {
Productid int
Quantity float64
}
db.Raw(`SELECT productid, SUM(quantity) AS quantity
FROM pos_order_items WHERE posorderid IN ?
GROUP BY productid`, billIDs).Scan(&consumed)
for _, c := range consumed {
qty := int(c.Quantity)
if float64(qty) < c.Quantity {
qty++ // the ingest rounds up, so the reversal must too
}
if err := db.Exec(`
INSERT INTO productstocks (tenantid, stockdate, locationid, productid,
quantity, stocktype, status)
VALUES (?, NOW(), ?, ?, ?, 'in', 'Active')`,
tenantID, locationID, c.Productid, qty).Error; err != nil {
fmt.Println(" return stock:", err)
continue
}
fmt.Printf(" returned %d unit(s) of product %d\n", qty, c.Productid)
}
db.Exec(`DELETE FROM pos_order_items WHERE posorderid IN ?`, billIDs)
db.Exec(`DELETE FROM pos_orders WHERE posorderid IN ?`, billIDs)
fmt.Printf(" deleted %d bill(s) and their items\n", len(billIDs))
var left int64
db.Raw(`SELECT COUNT(*) FROM pos_orders WHERE terminalorderid = ?`, testOrder).Scan(&left)
fmt.Printf("\nremaining test rows: %d\n", left)
var stock float64
db.Raw(`SELECT COALESCE(SUM(CASE WHEN LOWER(stocktype)='in' THEN quantity ELSE 0 END) -
SUM(CASE WHEN LOWER(stocktype)='out' THEN quantity ELSE 0 END), 0)
FROM productstocks WHERE productid = 6988 AND locationid = ? AND tenantid = ?`,
locationID, tenantID).Scan(&stock)
fmt.Printf("Mysore Banana stock now: %.0f (was 750 before any probe)\n", stock)
}

16
services/posService.go
View File

@@ -19,6 +19,10 @@ type PosService interface {
TerminalHealth(ctx context.Context, terminalID string) (map[string]string, error)
LocationHealth(ctx context.Context, locationID string) ([]map[string]string, error)
Sales(f models.PosSalesFilter) (*models.PosSalesPage, error)
SaleDetail(locationID int, reference string) (*models.PosOrders, error)
SalesSummary(f models.PosSalesFilter) (*models.PosSalesSummary, error)
}
type posService struct {
@@ -53,3 +57,15 @@ func (s *posService) IngestCustomers(batch models.PosCustomerBatch) (*models.Pos
func (s *posService) Catalogue(storeID, since string, page, pageSize int) (*models.PosCatalogueResponse, error) {
return s.repo.Catalogue(storeID, since, page, pageSize)
}
func (s *posService) Sales(f models.PosSalesFilter) (*models.PosSalesPage, error) {
return s.repo.Sales(f)
}
func (s *posService) SaleDetail(locationID int, reference string) (*models.PosOrders, error) {
return s.repo.SaleDetail(locationID, reference)
}
func (s *posService) SalesSummary(f models.PosSalesFilter) (*models.PosSalesSummary, error) {
return s.repo.SalesSummary(f)
}

6
services/productService.go
View File

@@ -293,6 +293,11 @@ func (s *productService) ImportCatalogueProduct(reqs []models.ImportCataloguePro
}
}
// Price carries the selling price onto the per-store row. Omitting it
// left productlocations.price at 0 for every imported product, and that
// column — not products.retailprice — is what the store catalogue, the
// customer app and each order line read. The result was a catalogue
// where nothing had a price and every order booked an amount of 0.
locations = append(locations, models.Productlocations{
Tenantid: req.Tenantid,
Locationid: req.Locationid,
@@ -300,6 +305,7 @@ func (s *productService) ImportCatalogueProduct(reqs []models.ImportCataloguePro
Quantity: req.Quantity,
Stocktype: req.Stocktype,
Status: req.Status,
Price: float32(req.Retailprice),
})
}