engineering/make_post_sell
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases 2 Packages Wiki Activity Actions

feat: bounded SSE feeds for offer & auction state machines

Browse source 
New text/event-stream endpoints — /o/{offer_id}/events (buyer/seller only)
and /a/{auction_id}/events (public). Each polls the row ~every 1.5s, emits
a `data: {json}` frame on connect and whenever the state-machine state
changes, sends a heartbeat comment, then closes after ~25s so the browser
EventSource reconnects — "bounded" because uWSGI is sync (~16 worker
threads) and a long-lived SSE would starve the pool. Shared helper
lib/sse.py (sse_response / event_stream); it uses its own short-lived DB
session per poll (request.dbsession is already closed by pyramid_tm by the
time the streaming generator runs). Timings come from settings
(app.sse.hold_seconds / app.sse.poll_interval_seconds; test.ini sets them
tiny so the streaming tests finish in ~0.06s).

Client: auction.js opens the EventSource and feeds each frame into its
existing applyState(); it falls back to polling /a/{id}.json every 5s
where EventSource is unavailable. offer.js opens the EventSource on the
offer page and reload()s on a state change (the whole layout depends on
state / can_act). offer.j2 carries data-offer-state. Caddy auto-detects
text/event-stream and stops buffering — no Salt change.

Tests: 4 new functional tests (both endpoints stream the right
content-type + a state frame; 404 for outsiders / unknown ids). 994 passed.
This commit is contained in:
russell@unturf.com 2026-05-12 21:03:50 -04:00
parent 93c0facbbc
commit 157e6f769a
No known key found for this signature in database
12 changed files with 344 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

19
CLAUDE.md
View file

@ -450,10 +450,25 @@ Form sections:
offer_min_buyer_account_age_hours).
Routes (registered before `product_slug` / `shop_slug` catch-alls):
- `/a/{auction_id}` + `/a/{id}.json` + `/a/{id}/{bid,buy-now,watch,checkout}`
- `/p/{product_id}/offer` (open) + `/o/{offer_id}` + `/o/{id}/{counter,accept,decline,withdraw,checkout}`
- `/a/{auction_id}` + `/a/{id}.json` + `/a/{id}/{events,bid,buy-now,watch,checkout}`
- `/p/{product_id}/offer` (open) + `/o/{offer_id}` + `/o/{id}/{events,counter,accept,decline,withdraw,checkout}`
- `/s/{shop_id}/offers` — operator inbox (`@shop_editor_required`), linked from `/actions/view`
Live updates use **bounded SSE** (`lib/sse.py` → `sse_response` /
`event_stream`): `/o/{id}/events` (buyer/seller only) and `/a/{id}/events`
(public) stream `text/event-stream`, poll the row ~every 1.5s, emit on
state change, heartbeat, then close after ~25s so `EventSource`
reconnects. uWSGI is sync (~16 threads) so a truly long-lived SSE would
starve the pool — hence "bounded". The generator must use its **own**
short-lived DB session per poll (`request.registry["dbsession_factory"]`),
**not** `request.dbsession` (pyramid_tm has already closed it by the time
the streaming generator runs). Timings: `app.sse.hold_seconds` /
`app.sse.poll_interval_seconds` settings (tiny in `test.ini`). `offer.js`
reloads on a state change; `auction.js` calls `applyState()` per frame and
falls back to 5s polling of `/a/{id}.json` where `EventSource` is absent.
Caddy auto-detects `text/event-stream` and stops buffering — no Salt
change needed.
Offer/auction POST routes are **capability-driven**: a plain browser
submit gets a flash + `302` redirect; an AJAX submit (`X-Requested-With:
XMLHttpRequest`) gets JSON. `static/js/offer.js` + `auction.js` are the

12
docs/auction-house.md
View file

@ -78,7 +78,8 @@ window means 1-min granularity is fine.
```
GET /a/{auction_id} live page (auction.j2 + auction.js)
GET /a/{auction_id}.json JSON state for poll
GET /a/{auction_id}.json JSON state (one-shot; fallback poll)
GET /a/{auction_id}/events bounded SSE feed of auction state (public)
POST /a/{auction_id}/bid place bid (login required, no self-bid)
POST /a/{auction_id}/buy-now end auction at buy_now (mode 2)
POST /a/{auction_id}/watch toggle watcher
@ -96,7 +97,14 @@ gift-card-purchases. After standard cart payment success,
## Live UI (`static/js/auction.js`)
- Countdown clock ticks every 1s (data-end-timestamp attribute)
- Polls `/a/{id}.json` every 5s for state changes
- Live state via a **bounded SSE feed** `/a/{id}/events` (`auction_events`
view → `lib/sse.py`): polls the row ~every 1.5s, emits a `data:` frame
on connect and on bid/soft-close/state change, heartbeats, then closes
after ~25s so `EventSource` reconnects (uWSGI sync workers can't hold
long-lived connections). `auction.js` calls `applyState()` per frame.
Where `EventSource` is unavailable it falls back to polling
`/a/{id}.json` every 5s. The countdown is NOT part of the SSE change
signal — the client derives it from `end_timestamp`.
- AJAX bid submit; success/error flash without page reload
The page works fully without JS (capability-driven presentation): `bid`,

18
docs/make-offer.md
View file

@ -112,8 +112,26 @@ POST /o/{offer_id}/decline decline current amount (terminal)
POST /o/{offer_id}/withdraw buyer-only terminal pull
POST /o/{offer_id}/checkout buyer pays accepted offer
GET /s/{shop_id}/offers operator inbox of all offers for the shop
GET /o/{offer_id}/events bounded SSE feed of the offer's state (buyer/seller only)
```
### Live updates (bounded SSE)
`/o/{offer_id}/events` is a `text/event-stream` that polls the offer row
every ~1.5s, emits a `data: {json}` frame on connect and whenever the
state-machine state changes, sends a heartbeat comment, then closes after
~25s — the browser `EventSource` reconnects. This caps the worker-thread
hold per client (uWSGI is sync, ~16 threads; a truly long-lived SSE would
starve the pool). Shared helper: `lib/sse.py` (`sse_response` /
`event_stream`). Timings are settings (`app.sse.hold_seconds`,
`app.sse.poll_interval_seconds`; `test.ini` sets them tiny). The
generator uses its **own** short-lived DB session per poll (not
`request.dbsession`, which pyramid_tm has already closed by then).
`offer.js` opens the `EventSource` on the offer page and `reload()`s on a
state change (the whole page layout depends on state/`can_act`). Auctions
have the same: `/a/{auction_id}/events` (public) + `auction.js` calls
`applyState()` on each frame.
`offer_open` is registered before the `product_slug` catch-all so
`/p/{id}/offer` is not shadowed.

76
make_post_sell/lib/sse.py Normal file
View file

@ -0,0 +1,76 @@
"""Bounded server-sent-events helper (MPS-20 / MPS-21).
We don't have Redis/pub-sub (SQLite app) and uWSGI only has a handful of
worker threads, so a truly long-lived SSE connection would starve the
pool. Instead each connection is *bounded*: it polls the row every
~`poll_interval` seconds, emits a `data:` event whenever the serialized
state changes (and the current state immediately on connect), sends a
heartbeat comment periodically, then closes after ~`hold_seconds`. The
browser's ``EventSource`` reconnects automatically, so the worker thread
is only held for that bounded window.
`fetch_state(session)` is invoked each poll with a *fresh* SQLAlchemy
session (NOT `request.dbsession` — by the time this generator runs,
pyramid_tm has already closed the request transaction) and must return
either ``(version, payload_dict)`` or ``None`` if the object is gone.
`version` is any comparable value; when it changes we emit `payload_dict`.
Timings come from settings so tests can run fast:
app.sse.hold_seconds (default 25)
app.sse.poll_interval_seconds (default 1.5)
"""
import json
import time
def _float_setting(settings, key, default):
try:
return float(settings.get(key, default))
except (TypeError, ValueError):
return default
def event_stream(request, fetch_state):
settings = request.registry.settings
hold = _float_setting(settings, "app.sse.hold_seconds", 25.0)
poll = _float_setting(settings, "app.sse.poll_interval_seconds", 1.5)
heartbeat_every = max(1, int(round(15.0 / poll)) if poll else 1)
factory = request.registry["dbsession_factory"]
deadline = time.monotonic() + hold
sentinel = object()
last_version = sentinel
i = 0
while time.monotonic() < deadline:
session = factory()
try:
result = fetch_state(session)
finally:
session.close()
if result is None:
yield b"event: gone\ndata: {}\n\n"
return
version, payload = result
if version != last_version:
last_version = version
yield ("data: " + json.dumps(payload) + "\n\n").encode("utf-8")
elif i % heartbeat_every == 0:
yield b": ping\n\n"
i += 1
if time.monotonic() >= deadline:
break
time.sleep(poll)
# Closing on purpose — EventSource will reconnect.
yield b"event: reconnect\ndata: {}\n\n"
def sse_response(request, fetch_state):
"""Attach an `event_stream` generator to `request.response` and return it."""
response = request.response
response.content_type = "text/event-stream"
response.headers["Cache-Control"] = "no-cache"
# Hint proxies (nginx; Caddy auto-detects text/event-stream) not to buffer.
response.headers["X-Accel-Buffering"] = "no"
response.app_iter = event_stream(request, fetch_state)
return response

2
make_post_sell/routes.py
View file

@ -243,6 +243,7 @@ def includeme(config):
# /a/{id}.json matches the JSON view; otherwise {auction_id} would
# greedily match "abc.json" and shadow the JSON endpoint.
config.add_route("auction_json", "/a/{auction_id}.json")
config.add_route("auction_events", "/a/{auction_id}/events")
config.add_route("auction_bid", "/a/{auction_id}/bid")
config.add_route("auction_buy_now", "/a/{auction_id}/buy-now")
config.add_route("auction_watch", "/a/{auction_id}/watch")
@ -251,6 +252,7 @@ def includeme(config):
# Offers (MPS-21). offer_open is registered earlier near product routes
# to avoid shadow by product_slug. /o/{id}/* sub-paths first, bare last.
config.add_route("offer_events", "/o/{offer_id}/events")
config.add_route("offer_counter", "/o/{offer_id}/counter")
config.add_route("offer_accept", "/o/{offer_id}/accept")
config.add_route("offer_decline", "/o/{offer_id}/decline")

24
make_post_sell/static/js/auction.js
View file

@ -114,8 +114,28 @@
});
}
// Start ticking every second; poll every 5 seconds.
// Countdown ticks locally every second.
setInterval(tickCountdown, 1000);
tickCountdown();
pollInterval = setInterval(poll, 5000);
// Live state: prefer a bounded SSE feed (the server closes it after
// ~25s; EventSource reconnects on its own), and fall back to polling
// /a/{id}.json every 5s where EventSource isn't available.
if (window.EventSource) {
var es = new EventSource("/a/" + auctionId + "/events");
es.onmessage = function (e) {
try {
applyState(JSON.parse(e.data));
} catch (err) {
/* ignore malformed frame */
}
};
es.addEventListener("gone", function () {
es.close();
});
// 'reconnect' frames and transport errors are handled by
// EventSource's own auto-reconnect — nothing to do here.
} else {
pollInterval = setInterval(poll, 5000);
}
})();

43
make_post_sell/static/js/offer.js
View file

@ -86,9 +86,46 @@
});
}
if (document.readyState === "loading") {
document.addEventListener("DOMContentLoaded", wire);
} else {
// Live state on the offer detail page via a bounded SSE feed. When the
// offer's state changes (counter / accept / decline / expire / pay) the
// whole page layout changes (which forms show, the notice, can_act), so
// the correct response is a full reload. EventSource auto-reconnects
// when the bounded stream closes.
function watchOfferState() {
var page = document.querySelector(".offer-page[data-offer-id]");
if (!page || !window.EventSource) return;
var offerId = page.getAttribute("data-offer-id");
var lastState = parseInt(page.getAttribute("data-offer-state"), 10);
var es = new EventSource("/o/" + offerId + "/events");
es.onmessage = function (e) {
var s;
try {
s = JSON.parse(e.data);
} catch (err) {
return;
}
if (typeof s.state !== "number") return;
if (isNaN(lastState)) {
lastState = s.state;
return;
}
if (s.state !== lastState) {
window.location.reload();
}
};
es.addEventListener("gone", function () {
es.close();
});
}
function init() {
wire();
watchOfferState();
}
if (document.readyState === "loading") {
document.addEventListener("DOMContentLoaded", init);
} else {
init();
}
})();

2
make_post_sell/templates/offer.j2
View file

@ -5,7 +5,7 @@
{%- endblock %}
{% block content %}
<section class="one-column offer-page" data-offer-id="{{ id }}">
<section class="one-column offer-page" data-offer-id="{{ id }}" data-offer-state="{{ state }}">
<div class="offer-header well">
<h1 class="type-title">Offer for "{{ product_title }}"</h1>

53
make_post_sell/tests/test_functional.py
View file

@ -5710,6 +5710,28 @@ class TestAuctionRoutes(_AuthenticatedBase):
self.assertEqual(auction.state, AUCTION_STATE_ENDED)
self.assertEqual(auction.winner, u2)
def test_auction_events_streams_state(self):
auction_id = self._make_active_auction(start_price=1000)
self.testapp.get("/log-out") # public — no login needed
sse = self.testapp.get(f"/a/{auction_id}/events", status=200)
self.assertIn("text/event-stream", sse.headers["Content-Type"])
body = sse.body.decode()
self.assertIn("data:", body)
import json as _json
first = [
ln for ln in body.splitlines() if ln.startswith("data: {")
][0]
payload = _json.loads(first[len("data: "):])
self.assertEqual(payload["id"], auction_id)
self.assertIn("current_high_in_cents", payload)
self.assertIn("bid_count", payload)
def test_auction_events_404_unknown(self):
self.testapp.get(
"/a/00000000000000000000000000000000/events",
expect_errors=True, status=404,
)
class TestAuctionNoJsFallback(_AuthenticatedBase):
"""MPS-20 capability-driven presentation: every auction action works
@ -5971,6 +5993,37 @@ class TestOfferRoutes(_AuthenticatedBase):
)
self.assertEqual(res2.status_int, 403)
def test_offer_events_streams_state(self):
product_id = self._make_offer_product(list_price=10000)
self.testapp.get("/log-out")
self.log_in_user(self.user2_creds)
res = self._ajax_post(f"/p/{product_id}/offer", {"amount": "70.00"})
offer_id = res.json["offer_id"]
# Buyer may subscribe to their own offer's SSE feed.
sse = self.testapp.get(f"/o/{offer_id}/events", status=200)
self.assertIn("text/event-stream", sse.headers["Content-Type"])
body = sse.body.decode()
self.assertIn("data:", body)
import json as _json
first = [
ln for ln in body.splitlines() if ln.startswith("data: {")
][0]
payload = _json.loads(first[len("data: "):])
self.assertEqual(payload["id"], offer_id)
self.assertIn("state", payload)
def test_offer_events_404_for_outsider(self):
product_id = self._make_offer_product(list_price=10000)
self.testapp.get("/log-out")
self.log_in_user(self.user2_creds)
res = self._ajax_post(f"/p/{product_id}/offer", {"amount": "70.00"})
offer_id = res.json["offer_id"]
# Anonymous (and any non-buyer/non-seller) gets 404 — offers are private.
self.testapp.get("/log-out")
self.testapp.get(
f"/o/{offer_id}/events", expect_errors=True, status=404
)
def test_terminal_action_forms_have_confirm_prompt(self):
product_id = self._make_offer_product(list_price=10000)
self.testapp.get("/log-out")

50
make_post_sell/views/auction.py
View file

@ -116,6 +116,56 @@ def auction_json(request):
return _serialize_auction(auction, request)
@view_config(route_name="auction_events")
def auction_events(request):
"""Bounded SSE feed of an auction's state (public — anyone may watch).
Emits on bid / soft-close extension / state transition. The countdown
is NOT part of the change signal (it would fire every poll) — the
client derives it from `end_timestamp` locally. EventSource reconnects
when the bounded stream closes.
"""
from ..lib.sse import sse_response
auction = get_auction_by_id(
request.dbsession, request.matchdict["auction_id"]
)
if auction is None:
raise HTTPNotFound()
auction_id = auction.uuid_str
def fetch_state(session):
a = get_auction_by_id(session, auction_id)
if a is None:
return None
bid_count = a.bids.count()
version = (
a.state, a.current_high_in_cents, a.end_timestamp, bid_count,
)
payload = {
"id": auction_id,
"state": a.state,
"state_human": a.state_human,
"is_active": a.is_active,
"is_ended": a.is_ended,
"is_terminal": a.is_terminal,
"current_high_in_cents": a.current_high_in_cents,
"current_high": a.current_high,
"min_next_bid_in_cents": a.min_next_bid_in_cents,
"min_next_bid": a.min_next_bid,
"end_timestamp": a.end_timestamp,
"time_remaining_ms": a.time_remaining_ms,
"bid_count": bid_count,
"has_reserve": a.has_reserve,
"reserve_met": a.reserve_met,
"winner_user_id": a.winner.uuid_str if a.winner else None,
}
return version, payload
return sse_response(request, fetch_state)
@view_config(route_name="auction_bid", request_method="POST", renderer="json")
@user_required(flash_msg="Please log in to bid.")
def auction_bid(request):

49
make_post_sell/views/offer.py
View file

@ -240,6 +240,55 @@ def offer_page(request):
return ctx
@view_config(route_name="offer_events")
def offer_events(request):
"""Bounded SSE feed of an offer's state-machine state.
Only the buyer or a shop owner may subscribe (offers are private).
The browser EventSource reconnects when the bounded stream closes.
"""
from ..lib.sse import sse_response
offer = get_offer_by_id(request.dbsession, request.matchdict["offer_id"])
if offer is None:
raise HTTPNotFound()
_party, can_view = _user_party(request, offer)
if not can_view:
raise HTTPNotFound()
offer_id = offer.uuid_str
def fetch_state(session):
o = get_offer_by_id(session, offer_id)
if o is None:
return None
version = (
o.state, o.current_amount_in_cents, o.round_count,
o.last_action_timestamp,
)
payload = {
"id": offer_id,
"state": o.state,
"state_human": o.state_human,
"current_amount_in_cents": o.current_amount_in_cents,
"current_amount": o.current_amount,
"round_count": o.round_count,
"current_party": o.current_party,
"is_open": o.is_open,
"is_terminal": o.is_terminal,
"is_pending": o.is_pending,
"is_countered": o.is_countered,
"is_accepted": o.is_accepted,
"is_expired": o.is_expired,
"is_declined": o.state == OFFER_STATE_DECLINED,
"is_withdrawn": o.state == OFFER_STATE_WITHDRAWN,
"is_paid": o.is_paid,
}
return version, payload
return sse_response(request, fetch_state)
def _offer_action(request, action_fn, message_required=False):
"""Common handling for counter / accept / decline / withdraw.

6
test.ini
View file

@ -41,6 +41,12 @@ app.payments.paypal.enabled = True
app.features.karaoke.enabled = True
app.features.torrent.enabled = True
# Bounded SSE (offer/auction live feeds) — tiny windows so the streaming
# endpoints return almost immediately under webtest instead of holding for
# ~25s. Production / development use the code defaults (25s hold, 1.5s poll).
app.sse.hold_seconds = 0.05
app.sse.poll_interval_seconds = 0.02
# PayPal sandbox mode for testing
app.paypal.sandbox_mode = True