Merge pull request 'feat(sidecar): store and serve the player-vendor market index' (#19) from feat/vendor-listing into edge
Reviewed-on: #19
This commit is contained in:
@@ -214,6 +214,44 @@ async fn main() -> anyhow::Result<()> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
// Player-vendor market index (Protocol 3.0). Each frame is authoritative for
|
||||||
|
// one vendor — the shard's round-robin sweep only emits a shop whose contents,
|
||||||
|
// prices or location actually moved — so this is a whole-row overwrite.
|
||||||
|
//
|
||||||
|
// Unlike the boards above there IS a remove: a vendor is dismissed, expires, or
|
||||||
|
// its owner switches off the in-game Vendor Search flag, and any of those must
|
||||||
|
// take the shop off the site. The last of the three is a privacy control, so
|
||||||
|
// dropping the row promptly is the point rather than housekeeping.
|
||||||
|
"vendor.listing" => {
|
||||||
|
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||||
|
let loc = ev.value.get("location");
|
||||||
|
let field = |k: &str| loc.and_then(|l| l.get(k));
|
||||||
|
if let Err(e) = event_store
|
||||||
|
.upsert_vendor(
|
||||||
|
serial,
|
||||||
|
ev.value.get("shopName").and_then(|v| v.as_str()),
|
||||||
|
ev.value.get("ownerName").and_then(|v| v.as_str()),
|
||||||
|
field("map").and_then(|v| v.as_str()),
|
||||||
|
field("x").and_then(|v| v.as_i64()),
|
||||||
|
field("y").and_then(|v| v.as_i64()),
|
||||||
|
field("region").and_then(|v| v.as_str()),
|
||||||
|
ev.value.get("count").and_then(|v| v.as_i64()),
|
||||||
|
&text,
|
||||||
|
t,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
tracing::warn!(error = %e, "failed to upsert vendor listing");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
"vendor.listing.remove" => {
|
||||||
|
if let Some(serial) = ev.value.get("serial").and_then(|s| s.as_str()) {
|
||||||
|
if let Err(e) = event_store.delete_vendor(serial).await {
|
||||||
|
tracing::warn!(error = %e, "failed to remove vendor listing");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
// Shard ruleset (Protocol 3.0): a singleton projection. The shard re-emits
|
// Shard ruleset (Protocol 3.0): a singleton projection. The shard re-emits
|
||||||
// world.ruleset on every connect, so this row is simply overwritten; `rev`
|
// world.ruleset on every connect, so this row is simply overwritten; `rev`
|
||||||
// lets a reader tell a re-send from an actual config change.
|
// lets a reader tell a re-send from an actual config change.
|
||||||
|
|||||||
@@ -371,6 +371,91 @@ impl Store {
|
|||||||
Ok(row.and_then(|r| serde_json::from_str(&r.get::<String, _>("json")).ok()))
|
Ok(row.and_then(|r| serde_json::from_str(&r.get::<String, _>("json")).ok()))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---- player-vendor market index (Protocol 3.0) ----
|
||||||
|
|
||||||
|
/// Upserts one vendor's whole listing, keyed by serial. Fed from `vendor.listing`, which the
|
||||||
|
/// shard emits as an authoritative per-vendor frame — so this replaces the row outright rather
|
||||||
|
/// than merging anything.
|
||||||
|
///
|
||||||
|
/// The items ride inside `json` and are deliberately NOT normalized into a `vendor_items`
|
||||||
|
/// table. The sidecar's job for the market is outage resilience (`PROTOCOL_2.md` §12.2) — hand
|
||||||
|
/// the website back what the shard last said — not search. Search lives in MariaDB on the
|
||||||
|
/// website side, where the query surface, the indexes and the cliloc-resolved display names
|
||||||
|
/// already are; a second search implementation here would be one more thing to keep in step
|
||||||
|
/// with it for no reader.
|
||||||
|
#[allow(clippy::too_many_arguments)]
|
||||||
|
pub async fn upsert_vendor(
|
||||||
|
&self,
|
||||||
|
serial: &str,
|
||||||
|
shop_name: Option<&str>,
|
||||||
|
owner_name: Option<&str>,
|
||||||
|
map: Option<&str>,
|
||||||
|
x: Option<i64>,
|
||||||
|
y: Option<i64>,
|
||||||
|
region: Option<&str>,
|
||||||
|
count: Option<i64>,
|
||||||
|
json: &str,
|
||||||
|
t: i64,
|
||||||
|
) -> anyhow::Result<()> {
|
||||||
|
sqlx::query(
|
||||||
|
"INSERT INTO vendors (serial, shop_name, owner_name, map, x, y, region, count, json, updated_t)
|
||||||
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
ON CONFLICT(serial) DO UPDATE SET shop_name = excluded.shop_name,
|
||||||
|
owner_name = excluded.owner_name, map = excluded.map, x = excluded.x, y = excluded.y,
|
||||||
|
region = excluded.region, count = excluded.count, json = excluded.json,
|
||||||
|
updated_t = excluded.updated_t",
|
||||||
|
)
|
||||||
|
.bind(serial)
|
||||||
|
.bind(shop_name)
|
||||||
|
.bind(owner_name)
|
||||||
|
.bind(map)
|
||||||
|
.bind(x)
|
||||||
|
.bind(y)
|
||||||
|
.bind(region)
|
||||||
|
.bind(count)
|
||||||
|
.bind(json)
|
||||||
|
.bind(t)
|
||||||
|
.execute(&self.pool)
|
||||||
|
.await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Drops one vendor from the index. Fed from `vendor.listing.remove` — a vendor dismissed,
|
||||||
|
/// expired, or whose owner switched off its in-game Vendor Search flag.
|
||||||
|
pub async fn delete_vendor(&self, serial: &str) -> anyhow::Result<()> {
|
||||||
|
sqlx::query("DELETE FROM vendors WHERE serial = ?")
|
||||||
|
.bind(serial)
|
||||||
|
.execute(&self.pool)
|
||||||
|
.await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// One page of the index, ordered by serial.
|
||||||
|
///
|
||||||
|
/// Paged where the other boards are not, and the ordering is why it can be: a whole-world
|
||||||
|
/// market is the one board that does not fit in a response. Ordering by SERIAL rather than by
|
||||||
|
/// shop name is deliberate — the page is a snapshot cursor for the website's reconnect
|
||||||
|
/// backfill, and a serial is stable while a shop name is renameable, so a rename mid-backfill
|
||||||
|
/// cannot make a vendor skip or repeat a page.
|
||||||
|
pub async fn vendors_page(&self, limit: i64, offset: i64) -> anyhow::Result<Vec<Value>> {
|
||||||
|
let limit = limit.clamp(1, 1000);
|
||||||
|
let offset = offset.max(0);
|
||||||
|
let rows = sqlx::query("SELECT json FROM vendors ORDER BY serial LIMIT ? OFFSET ?")
|
||||||
|
.bind(limit)
|
||||||
|
.bind(offset)
|
||||||
|
.fetch_all(&self.pool)
|
||||||
|
.await?;
|
||||||
|
Ok(parse_json_column(rows))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// How many vendors the index holds, so a paging caller knows when to stop.
|
||||||
|
pub async fn vendors_count(&self) -> anyhow::Result<i64> {
|
||||||
|
let row = sqlx::query("SELECT COUNT(*) AS n FROM vendors")
|
||||||
|
.fetch_one(&self.pool)
|
||||||
|
.await?;
|
||||||
|
Ok(row.get::<i64, _>("n"))
|
||||||
|
}
|
||||||
|
|
||||||
// ---- Town Cryer news (Protocol 2.1) ----
|
// ---- Town Cryer news (Protocol 2.1) ----
|
||||||
|
|
||||||
/// Stores/replaces one external news article (the `news.add` command json), keyed by id. The
|
/// Stores/replaces one external news article (the `news.add` command json), keyed by id. The
|
||||||
@@ -480,6 +565,23 @@ CREATE TABLE IF NOT EXISTS points_boards (
|
|||||||
updated_t INTEGER NOT NULL
|
updated_t INTEGER NOT NULL
|
||||||
);
|
);
|
||||||
|
|
||||||
|
-- Player-vendor market index (Protocol 3.0). One row per vendor, holding the whole authoritative
|
||||||
|
-- `vendor.listing` frame including its items. The hoisted columns exist for the ORDER BY and for
|
||||||
|
-- an operator eyeballing the table; nothing here is searched, because search is the website's job
|
||||||
|
-- (see upsert_vendor). Rows are dropped on `vendor.listing.remove`.
|
||||||
|
CREATE TABLE IF NOT EXISTS vendors (
|
||||||
|
serial TEXT PRIMARY KEY,
|
||||||
|
shop_name TEXT,
|
||||||
|
owner_name TEXT,
|
||||||
|
map TEXT,
|
||||||
|
x INTEGER,
|
||||||
|
y INTEGER,
|
||||||
|
region TEXT,
|
||||||
|
count INTEGER,
|
||||||
|
json TEXT NOT NULL,
|
||||||
|
updated_t INTEGER NOT NULL
|
||||||
|
);
|
||||||
|
|
||||||
-- The shard's published ruleset (Protocol 3.0). Singleton: the CHECK is what makes it one,
|
-- The shard's published ruleset (Protocol 3.0). Singleton: the CHECK is what makes it one,
|
||||||
-- so an upsert can target id = 1 unconditionally and no second row can ever appear.
|
-- so an upsert can target id = 1 unconditionally and no second row can ever appear.
|
||||||
CREATE TABLE IF NOT EXISTS ruleset (
|
CREATE TABLE IF NOT EXISTS ruleset (
|
||||||
|
|||||||
@@ -90,6 +90,12 @@ pub async fn serve(addr: &str, state: AppState) -> anyhow::Result<()> {
|
|||||||
// whole set, or one system by its PointsType name.
|
// whole set, or one system by its PointsType name.
|
||||||
.route("/points", get(points))
|
.route("/points", get(points))
|
||||||
.route("/points/:system", get(points_system))
|
.route("/points/:system", get(points_system))
|
||||||
|
// The player-vendor market index (Protocol 3.0). `/market`, NOT `/vendors`: axum would
|
||||||
|
// route the latter fine, but `/vendors/:account` next door is the per-account RPC, and two
|
||||||
|
// routes a prefix apart that mean "this player's shops" and "every shop on the shard" is a
|
||||||
|
// readability trap nobody wins. The only PAGED read the sidecar serves — a whole-world
|
||||||
|
// market does not fit in one response.
|
||||||
|
.route("/market", get(market))
|
||||||
.route_layer(middleware::from_fn_with_state(state.clone(), gate));
|
.route_layer(middleware::from_fn_with_state(state.clone(), gate));
|
||||||
|
|
||||||
let app = Router::new()
|
let app = Router::new()
|
||||||
@@ -872,6 +878,54 @@ async fn online(State(st): State<AppState>) -> impl IntoResponse {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Deserialize)]
|
||||||
|
struct PageQuery {
|
||||||
|
limit: Option<i64>,
|
||||||
|
offset: Option<i64>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The player-vendor market index: every vendor's shop name, owner, location and priced inventory,
|
||||||
|
/// as the shard last published it. Store-backed like the other boards, which is what lets the
|
||||||
|
/// website's market page render (labelled stale) while the shard is down.
|
||||||
|
///
|
||||||
|
/// Paged — `?limit=&offset=`, limit clamped to 1..1000, default 200 — because this is the one board
|
||||||
|
/// that can be a whole world's inventory. `total` is returned alongside so the caller knows when to
|
||||||
|
/// stop rather than paging until it sees a short page, which would race a concurrent sweep.
|
||||||
|
///
|
||||||
|
/// The frames are served VERBATIM, including owner names and coordinates. That is not an oversight:
|
||||||
|
/// the sidecar defines no audiences (docs/link/v3.md §3.2). Deciding who may see a vendor's owner
|
||||||
|
/// or whereabouts is the website's job and is admin-configurable there.
|
||||||
|
async fn market(State(st): State<AppState>, Query(q): Query<PageQuery>) -> impl IntoResponse {
|
||||||
|
let limit = q.limit.unwrap_or(200);
|
||||||
|
let offset = q.offset.unwrap_or(0);
|
||||||
|
|
||||||
|
let total = match st.store.vendors_count().await {
|
||||||
|
Ok(n) => n,
|
||||||
|
Err(e) => {
|
||||||
|
return (
|
||||||
|
StatusCode::INTERNAL_SERVER_ERROR,
|
||||||
|
Json(json!({"error": e.to_string()})),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
match st.store.vendors_page(limit, offset).await {
|
||||||
|
Ok(vendors) => (
|
||||||
|
StatusCode::OK,
|
||||||
|
Json(json!({
|
||||||
|
"vendors": vendors,
|
||||||
|
"total": total,
|
||||||
|
"limit": limit.clamp(1, 1000),
|
||||||
|
"offset": offset.max(0),
|
||||||
|
})),
|
||||||
|
),
|
||||||
|
Err(e) => (
|
||||||
|
StatusCode::INTERNAL_SERVER_ERROR,
|
||||||
|
Json(json!({"error": e.to_string()})),
|
||||||
|
),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// ---- websocket ----
|
// ---- websocket ----
|
||||||
|
|
||||||
async fn ws_upgrade(ws: WebSocketUpgrade, State(state): State<AppState>) -> impl IntoResponse {
|
async fn ws_upgrade(ws: WebSocketUpgrade, State(state): State<AppState>) -> impl IntoResponse {
|
||||||
|
|||||||
Reference in New Issue
Block a user