Rewards beyond titles: bundles on quests, achievements and rank milestones; panel-side money/chunks/badges; delivery queue

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DjMbLQujBHunCCu5GpsHaT

panel/server/src/db.rs

@@ -719,6 +719,46 @@ const MIGRATIONS: &[&str] = &[
     );
     CREATE INDEX guild_join_requests_player ON guild_join_requests(uuid);
     "#,
+    // Rewards beyond XP and titles: what quests, achievements and rank milestones hand out (money, items, permissions,
+    // claim chunks, commands…), the queue of earned bundles, and the deliveries waiting for a game server.
+    r#"
+    CREATE TABLE reward_bundles (
+        source_type TEXT NOT NULL,
+        source_id TEXT NOT NULL,
+        actions TEXT NOT NULL DEFAULT '[]',
+        PRIMARY KEY (source_type, source_id)
+    );
+    CREATE TABLE reward_queue (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        uuid TEXT NOT NULL,
+        source_type TEXT NOT NULL,
+        source_id TEXT NOT NULL,
+        created_at TEXT NOT NULL
+    );
+    CREATE TABLE reward_deliveries (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        uuid TEXT NOT NULL,
+        server_id INTEGER REFERENCES game_servers(id) ON DELETE CASCADE,
+        kind TEXT NOT NULL,
+        payload TEXT NOT NULL DEFAULT '{}',
+        source TEXT NOT NULL DEFAULT '',
+        created_at TEXT NOT NULL,
+        sent_at TEXT,
+        delivered_at TEXT,
+        attempts INTEGER NOT NULL DEFAULT 0,
+        error TEXT
+    );
+    CREATE INDEX reward_deliveries_pending ON reward_deliveries(delivered_at, server_id);
+    CREATE TABLE player_bonuses (
+        uuid TEXT PRIMARY KEY,
+        claim_chunks INTEGER NOT NULL DEFAULT 0
+    );
+    CREATE TRIGGER reward_on_achievement AFTER INSERT ON user_achievements
+    WHEN EXISTS (SELECT 1 FROM reward_bundles WHERE source_type = 'achievement' AND source_id = NEW.achievement_id) BEGIN
+      INSERT INTO reward_queue (uuid, source_type, source_id, created_at)
+      VALUES (NEW.user_uuid, 'achievement', NEW.achievement_id, strftime('%Y-%m-%dT%H:%M:%SZ', 'now'));
+    END;
+    "#,
 ];
 
 pub async fn connect(data_dir: &Path) -> Result<SqlitePool> {

panel/server/src/lib.rs

@@ -12,6 +12,7 @@ pub mod net;
 pub mod packs;
 pub mod progression;
 pub mod purge;
+pub mod rewards;
 pub mod routes;
 pub mod seed;
 pub mod state;

panel/server/src/main.rs

@@ -23,6 +23,18 @@ async fn main() -> anyhow::Result<()> {
     let state = build_state(cfg, pool).await?;
     bootstrap_admin(&state).await?;
     scopenet_panel::routes::discord::spawn_worker(state.clone());
+    // Earned rewards are applied within seconds even when no game server happens to be polling.
+    {
+        let state = state.clone();
+        tokio::spawn(async move {
+            loop {
+                tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+                if let Err(e) = scopenet_panel::rewards::process_queue(&state).await {
+                    tracing::warn!("reward queue failed: {:?}", e.message);
+                }
+            }
+        });
+    }
 
     let listener = tokio::net::TcpListener::bind(&bind).await?;
     tracing::info!("SCOPENET panel v{} listening on http://{bind}", env!("CARGO_PKG_VERSION"));

panel/server/src/rewards.rs

@@ -0,0 +1,293 @@
+//! Rewards beyond XP and titles.
+//!
+//! An admin attaches a *bundle* of actions to a quest, an achievement or a rank milestone (a level reward). When a
+//! player earns one, the source goes into `reward_queue`; [`process_queue`] then applies each action: money, claim
+//! chunks and badges happen in the panel, while items, permissions, groups, commands and messages become
+//! `reward_deliveries` that the player's game server collects (once the player is online) and acknowledges.
+
+use crate::error::{AppError, AppResult};
+use crate::state::AppState;
+use serde_json::{json, Map, Value};
+use sqlx::SqliteConnection;
+
+pub const MAX_ACTIONS: usize = 20;
+const SOURCES: [&str; 3] = ["quest", "achievement", "level_reward"];
+
+pub fn check_source(kind: &str) -> AppResult<()> {
+    if SOURCES.contains(&kind) { Ok(()) } else { Err(AppError::bad_request("unknown reward source")) }
+}
+
+fn text(a: &Map<String, Value>, key: &str, max: usize) -> String {
+    a.get(key).and_then(Value::as_str).map(|s| s.trim().chars().filter(|c| !c.is_control()).take(max).collect()).unwrap_or_default()
+}
+
+fn int(a: &Map<String, Value>, key: &str) -> Option<i64> {
+    a.get(key).and_then(|v| v.as_i64().or_else(|| v.as_f64().map(|f| f as i64)))
+}
+
+fn word(value: &str, extra: &str) -> bool {
+    !value.is_empty() && value.chars().all(|c| c.is_ascii_alphanumeric() || extra.contains(c))
+}
+
+/// Validate and normalise an admin's list of actions. Unknown fields are dropped, so nothing unexpected is stored.
+pub fn validate(actions: &Value) -> AppResult<Vec<Value>> {
+    let list = actions.as_array().ok_or_else(|| AppError::bad_request("actions must be a list"))?;
+    if list.len() > MAX_ACTIONS {
+        return Err(AppError::bad_request(format!("at most {MAX_ACTIONS} rewards per bundle")));
+    }
+    let mut out = Vec::new();
+    for raw in list {
+        let a = raw.as_object().ok_or_else(|| AppError::bad_request("each reward must be an object"))?;
+        let kind = text(a, "type", 24);
+        let server = int(a, "server_id").filter(|s| *s > 0);
+        let mut v = Map::new();
+        v.insert("type".into(), json!(kind));
+        if let Some(s) = server {
+            v.insert("server_id".into(), json!(s));
+        }
+        match kind.as_str() {
+            "money" => {
+                let amount = a.get("amount").and_then(Value::as_f64).unwrap_or(0.0);
+                if !amount.is_finite() || amount <= 0.0 || amount > 1e9 {
+                    return Err(AppError::bad_request("money must be more than 0"));
+                }
+                v.insert("amount".into(), json!((amount * 100.0).round() / 100.0));
+            }
+            "item" => {
+                let mut item = text(a, "item", 100).to_lowercase();
+                if !item.contains(':') {
+                    item = format!("minecraft:{item}");
+                }
+                let (ns, path) = item.split_once(':').unwrap_or(("", ""));
+                if !word(ns, "_.-") || !word(path, "_.-/") {
+                    return Err(AppError::bad_request("item must look like minecraft:diamond"));
+                }
+                let amount = int(a, "amount").unwrap_or(1);
+                if !(1..=6400).contains(&amount) {
+                    return Err(AppError::bad_request("item amount must be 1 to 6400"));
+                }
+                v.insert("item".into(), json!(item));
+                v.insert("amount".into(), json!(amount));
+            }
+            "permission" => {
+                let node = text(a, "node", 100);
+                if !word(&node, "_.*-:") {
+                    return Err(AppError::bad_request("permission node has unusual characters"));
+                }
+                let minutes = int(a, "minutes").unwrap_or(0);
+                if !(0..=5_256_000).contains(&minutes) {
+                    return Err(AppError::bad_request("permission duration is out of range"));
+                }
+                v.insert("node".into(), json!(node));
+                v.insert("value".into(), json!(a.get("value").and_then(Value::as_bool).unwrap_or(true)));
+                v.insert("minutes".into(), json!(minutes));
+            }
+            "group" => {
+                let group = text(a, "group", 64);
+                if !word(&group, "_.-") {
+                    return Err(AppError::bad_request("group name has unusual characters"));
+                }
+                v.insert("group".into(), json!(group));
+            }
+            "claim_chunks" => {
+                let amount = int(a, "amount").unwrap_or(0);
+                if !(1..=10_000).contains(&amount) {
+                    return Err(AppError::bad_request("claim chunks must be 1 to 10000"));
+                }
+                v.insert("amount".into(), json!(amount));
+            }
+            "badge" => {
+                let badge = text(a, "badge", 40);
+                if badge.is_empty() {
+                    return Err(AppError::bad_request("badge needs a name"));
+                }
+                v.insert("badge".into(), json!(badge));
+            }
+            "message" => {
+                let message = text(a, "text", 200);
+                if message.is_empty() {
+                    return Err(AppError::bad_request("message can't be empty"));
+                }
+                v.insert("text".into(), json!(message));
+            }
+            "command" => {
+                let command = text(a, "command", 256).trim_start_matches('/').to_string();
+                if command.is_empty() {
+                    return Err(AppError::bad_request("command can't be empty"));
+                }
+                v.insert("command".into(), json!(command));
+            }
+            _ => return Err(AppError::bad_request(format!("unknown reward type '{kind}'"))),
+        }
+        out.push(Value::Object(v));
+    }
+    Ok(out)
+}
+
+fn pretty_item(id: &str) -> String {
+    let name = id.rsplit(':').next().unwrap_or(id).replace(['_', '/'], " ");
+    name.split_whitespace().map(|w| { let mut c = w.chars(); c.next().map(|f| f.to_uppercase().collect::<String>() + c.as_str()).unwrap_or_default() }).collect::<Vec<_>>().join(" ")
+}
+
+/// A short, player-facing line for one action (commands stay vague on purpose).
+pub fn describe(a: &Value) -> String {
+    let n = |k: &str| a[k].as_i64().unwrap_or(0);
+    match a["type"].as_str().unwrap_or("") {
+        "money" => format!("${:.2}", a["amount"].as_f64().unwrap_or(0.0)),
+        "item" => format!("{} x{}", pretty_item(a["item"].as_str().unwrap_or("")), n("amount")),
+        "permission" => {
+            let m = n("minutes");
+            let span = if m == 0 { String::new() } else if m % 1440 == 0 { format!(" for {} day{}", m / 1440, if m / 1440 == 1 { "" } else { "s" }) } else if m % 60 == 0 { format!(" for {}h", m / 60) } else { format!(" for {m}m") };
+            format!("Permission: {}{}", a["node"].as_str().unwrap_or(""), span)
+        }
+        "group" => format!("Group: {}", a["group"].as_str().unwrap_or("")),
+        "claim_chunks" => format!("+{} claim chunks", n("amount")),
+        "badge" => format!("Badge: {}", a["badge"].as_str().unwrap_or("")),
+        "message" => "A message".to_string(),
+        _ => "A special reward".to_string(),
+    }
+}
+
+pub async fn bundle(conn: &mut SqliteConnection, kind: &str, id: &str) -> AppResult<Vec<Value>> {
+    let raw: Option<String> = sqlx::query_scalar("SELECT actions FROM reward_bundles WHERE source_type = ? AND source_id = ?").bind(kind).bind(id).fetch_optional(&mut *conn).await?;
+    Ok(raw.and_then(|r| serde_json::from_str::<Vec<Value>>(&r).ok()).unwrap_or_default())
+}
+
+/// Queue a bundle for a player who just earned its source. Does nothing when the source has no bundle.
+pub async fn enqueue(conn: &mut SqliteConnection, uuid: &str, kind: &str, id: &str) -> AppResult<()> {
+    let has: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM reward_bundles WHERE source_type = ? AND source_id = ? AND actions <> '[]')").bind(kind).bind(id).fetch_one(&mut *conn).await?;
+    if has {
+        sqlx::query("INSERT INTO reward_queue (uuid, source_type, source_id, created_at) VALUES (?, ?, ?, ?)").bind(uuid).bind(kind).bind(id).bind(crate::db::now()).execute(&mut *conn).await?;
+    }
+    Ok(())
+}
+
+async fn credit(tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, uuid: &str, name: &str, amount: f64, server: Option<i64>, note: &str) -> AppResult<()> {
+    // One credit per economy: servers sharing a group share balances, so they must not be paid twice.
+    let servers: Vec<i64> = match server {
+        Some(id) => vec![id],
+        None => sqlx::query_scalar("SELECT id FROM game_servers ORDER BY id").fetch_all(&mut **tx).await?,
+    };
+    let mut economies = Vec::new();
+    for id in servers {
+        let e: i64 = sqlx::query_scalar(
+            "SELECT COALESCE((SELECT MIN(b.id) FROM game_servers b WHERE b.economy_group <> '' AND b.economy_group = a.economy_group COLLATE NOCASE), a.id) FROM game_servers a WHERE a.id = ?",
+        )
+        .bind(id)
+        .fetch_optional(&mut **tx)
+        .await?
+        .unwrap_or(id);
+        if !economies.contains(&e) {
+            economies.push(e);
+        }
+    }
+    let now = chrono::Utc::now().to_rfc3339();
+    for economy in economies {
+        crate::routes::economy::ensure_balance(tx, economy, uuid, name).await?;
+        sqlx::query("UPDATE server_economy SET balance = balance + ?, updated_at = ? WHERE server_id = ? AND uuid = ?").bind(amount).bind(&now).bind(economy).bind(uuid).execute(&mut **tx).await?;
+        sqlx::query("INSERT INTO economy_transactions(server_id, from_uuid, from_name, to_uuid, to_name, amount, description, created_at) VALUES (?, 'server', 'Rewards', ?, ?, ?, ?, ?)")
+            .bind(economy).bind(uuid).bind(name).bind(amount).bind(note).bind(&now).execute(&mut **tx).await?;
+    }
+    Ok(())
+}
+
+async fn source_label(conn: &mut SqliteConnection, kind: &str, id: &str) -> String {
+    let q = match kind {
+        "quest" => "SELECT title FROM quests WHERE id = ?",
+        "achievement" => "SELECT title FROM achievements WHERE id = ?",
+        _ => "SELECT reward_name FROM level_rewards WHERE CAST(id AS TEXT) = ?",
+    };
+    sqlx::query_scalar::<_, String>(q).bind(id).fetch_optional(&mut *conn).await.ok().flatten().unwrap_or_else(|| id.to_string())
+}
+
+/// Apply every queued bundle. Safe to call often; each queue row is handled exactly once.
+pub async fn process_queue(state: &AppState) -> AppResult<usize> {
+    let rows: Vec<(i64, String, String, String)> = sqlx::query_as("SELECT id, uuid, source_type, source_id FROM reward_queue ORDER BY id LIMIT 200").fetch_all(&state.db).await?;
+    let mut done = 0;
+    for (queue_id, uuid, kind, id) in rows {
+        let mut tx = state.db.begin().await?;
+        let gone: Option<i64> = sqlx::query_scalar("DELETE FROM reward_queue WHERE id = ? RETURNING id").bind(queue_id).fetch_optional(&mut *tx).await?;
+        if gone.is_none() {
+            continue; // another request got there first
+        }
+        let name: String = sqlx::query_scalar("SELECT username FROM users WHERE uuid = ?").bind(&uuid).fetch_optional(&mut *tx).await?.unwrap_or_default();
+        if name.is_empty() {
+            tx.commit().await?;
+            continue;
+        }
+        let label = source_label(&mut tx, &kind, &id).await;
+        let source = format!("{kind}:{id}");
+        let now = crate::db::now();
+        for action in bundle(&mut tx, &kind, &id).await? {
+            let server = action["server_id"].as_i64();
+            match action["type"].as_str().unwrap_or("") {
+                "money" => credit(&mut tx, &uuid, &name, action["amount"].as_f64().unwrap_or(0.0), server, &format!("Reward: {label}")).await?,
+                "claim_chunks" => {
+                    sqlx::query("INSERT INTO player_bonuses (uuid, claim_chunks) VALUES (?, ?) ON CONFLICT(uuid) DO UPDATE SET claim_chunks = claim_chunks + excluded.claim_chunks")
+                        .bind(&uuid).bind(action["amount"].as_i64().unwrap_or(0)).execute(&mut *tx).await?;
+                }
+                "badge" => {
+                    let raw: String = sqlx::query_scalar("SELECT badges FROM user_levels WHERE uuid = ?").bind(&uuid).fetch_optional(&mut *tx).await?.unwrap_or_else(|| "[]".into());
+                    let mut badges: Vec<String> = serde_json::from_str(&raw).unwrap_or_default();
+                    let badge = action["badge"].as_str().unwrap_or("").to_string();
+                    if !badge.is_empty() && !badges.contains(&badge) {
+                        badges.push(badge);
+                        sqlx::query("INSERT INTO user_levels(uuid, badges, updated_at) VALUES (?, ?, ?) ON CONFLICT(uuid) DO UPDATE SET badges = excluded.badges")
+                            .bind(&uuid).bind(serde_json::to_string(&badges)?).bind(&now).execute(&mut *tx).await?;
+                    }
+                }
+                other => {
+                    let mut payload = action.clone();
+                    payload["reason"] = json!(label);
+                    sqlx::query("INSERT INTO reward_deliveries (uuid, server_id, kind, payload, source, created_at) VALUES (?, ?, ?, ?, ?, ?)")
+                        .bind(&uuid).bind(server).bind(other).bind(payload.to_string()).bind(&source).bind(&now).execute(&mut *tx).await?;
+                }
+            }
+        }
+        tx.commit().await?;
+        done += 1;
+    }
+    Ok(done)
+}
+
+/// Deliveries this game server should carry out now: for players online here, not yet done, and not already out for delivery.
+pub async fn poll(state: &AppState, server_id: i64) -> AppResult<Vec<Value>> {
+    process_queue(state).await?;
+    let mut tx = state.db.begin().await?;
+    let stale = (chrono::Utc::now() - chrono::Duration::seconds(120)).to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
+    let rows: Vec<(i64, String, String, String, String)> = sqlx::query_as(
+        "SELECT d.id, d.uuid, COALESCE(u.username, ''), d.kind, d.payload FROM reward_deliveries d
+         LEFT JOIN users u ON u.uuid = d.uuid
+         WHERE d.delivered_at IS NULL AND d.attempts < 8 AND (d.server_id IS NULL OR d.server_id = ?)
+           AND (d.sent_at IS NULL OR d.sent_at < ?)
+           AND EXISTS (SELECT 1 FROM server_online o WHERE o.server_id = ? AND o.uuid = d.uuid)
+         ORDER BY d.id LIMIT 50",
+    )
+    .bind(server_id)
+    .bind(&stale)
+    .bind(server_id)
+    .fetch_all(&mut *tx)
+    .await?;
+    let now = crate::db::now();
+    for (id, ..) in &rows {
+        sqlx::query("UPDATE reward_deliveries SET sent_at = ?, attempts = attempts + 1 WHERE id = ?").bind(&now).bind(id).execute(&mut *tx).await?;
+    }
+    tx.commit().await?;
+    Ok(rows.into_iter().map(|(id, uuid, name, kind, payload)| json!({ "id": id, "uuid": uuid, "name": name, "kind": kind, "payload": serde_json::from_str::<Value>(&payload).unwrap_or(json!({})) })).collect())
+}
+
+/// The game server reports what it did. Failures are retried a few times, then kept in the log with the reason.
+pub async fn ack(state: &AppState, done: &[i64], failed: &[(i64, String)]) -> AppResult<()> {
+    let now = crate::db::now();
+    let mut tx = state.db.begin().await?;
+    for id in done {
+        sqlx::query("UPDATE reward_deliveries SET delivered_at = ?, error = NULL WHERE id = ?").bind(&now).bind(id).execute(&mut *tx).await?;
+    }
+    for (id, error) in failed {
+        let error: String = error.chars().filter(|c| !c.is_control()).take(200).collect();
+        // A refused delivery (e.g. commands switched off on that server) is retried a few times, then left for the admin to see.
+        sqlx::query("UPDATE reward_deliveries SET error = ?, sent_at = ? WHERE id = ? AND delivered_at IS NULL").bind(error).bind(&now).bind(id).execute(&mut *tx).await?;
+    }
+    tx.commit().await?;
+    Ok(())
+}

panel/server/src/routes/achievements.rs

@@ -311,6 +311,7 @@ pub async fn admin_delete_achievement(
     Path(id): Path<String>,
     State(state): State<AppState>,
 ) -> AppResult<Json<Value>> {
+    sqlx::query("DELETE FROM reward_bundles WHERE source_type = 'achievement' AND source_id = ?").bind(&id).execute(&state.db).await?;
     sqlx::query("DELETE FROM achievements WHERE id = ?")
         .bind(&id)
         .execute(&state.db)

panel/server/src/routes/guilds.rs

@@ -36,7 +36,7 @@ pub async fn list_guilds(Query(query): Query<InstanceQuery>, State(state): State
         i64,
     )> = sqlx::query_as(
         "SELECT g.id, g.instance_id, g.name, g.tag, g.description, g.motd, g.leader_uuid,
-                g.icon_url, g.banner_url, g.level, g.xp, g.max_claims, g.created_at,
+                g.icon_url, g.banner_url, g.level, g.xp, (g.max_claims + COALESCE((SELECT SUM(b.claim_chunks) FROM player_bonuses b JOIN guild_members m ON m.uuid = b.uuid WHERE m.guild_id = g.id), 0)), g.created_at,
                 (SELECT COUNT(*) FROM guild_members gm WHERE gm.guild_id = g.id) as member_count,
                 (SELECT COUNT(*) FROM guild_claims gc WHERE gc.guild_id = g.id) as claims_count
          FROM guilds g
@@ -214,7 +214,7 @@ async fn fetch_guild_detail(state: &AppState, guild_id: &str) -> AppResult<Guild
     let row: Option<(String, String, String, String, String, String, String, Option<String>, Option<String>, i64, i64, i64, String)> =
         sqlx::query_as(
             "SELECT id, instance_id, name, tag, description, motd, leader_uuid,
-                icon_url, banner_url, level, xp, max_claims, created_at
+                icon_url, banner_url, level, xp, (max_claims + COALESCE((SELECT SUM(b.claim_chunks) FROM player_bonuses b JOIN guild_members m ON m.uuid = b.uuid WHERE m.guild_id = guilds.id), 0)), created_at
          FROM guilds WHERE id = ?",
         )
         .bind(guild_id)
@@ -740,7 +740,7 @@ pub async fn claim_chunk(
         return Err(AppError::forbidden("Your guild role cannot claim land"));
     }
 
-    let max_claims: i64 = sqlx::query_scalar("SELECT max_claims FROM guilds WHERE id = ?").bind(&guild_id).fetch_one(&state.db).await?;
+    let max_claims: i64 = sqlx::query_scalar("SELECT (max_claims + COALESCE((SELECT SUM(b.claim_chunks) FROM player_bonuses b JOIN guild_members m ON m.uuid = b.uuid WHERE m.guild_id = guilds.id), 0)) FROM guilds WHERE id = ?").bind(&guild_id).fetch_one(&state.db).await?;
 
     let current_claims: i64 =
         sqlx::query_scalar("SELECT COUNT(*) FROM guild_claims WHERE guild_id = ?").bind(&guild_id).fetch_one(&state.db).await?;
@@ -932,7 +932,7 @@ pub async fn admin_list_guilds(_admin: AdminUser, State(state): State<AppState>)
         i64,
     )> = sqlx::query_as(
         "SELECT g.id, g.instance_id, g.name, g.tag, g.description, g.motd, g.leader_uuid,
-                g.icon_url, g.banner_url, g.level, g.xp, g.max_claims, g.created_at,
+                g.icon_url, g.banner_url, g.level, g.xp, (g.max_claims + COALESCE((SELECT SUM(b.claim_chunks) FROM player_bonuses b JOIN guild_members m ON m.uuid = b.uuid WHERE m.guild_id = g.id), 0)), g.created_at,
                 (SELECT COUNT(*) FROM guild_members gm WHERE gm.guild_id = g.id) as member_count,
                 (SELECT COUNT(*) FROM guild_claims gc WHERE gc.guild_id = g.id) as claims_count
          FROM guilds g
@@ -1150,7 +1150,7 @@ pub async fn server_claim_chunk(
     if !guild_can(&state, &guild_id, &payload.uuid, "claim").await? { return Err(AppError::forbidden("Your guild role cannot claim land")); }
 
     let max_claims: i64 =
-        sqlx::query_scalar("SELECT max_claims FROM guilds WHERE id = ?").bind(&guild_id).fetch_one(&state.db).await.unwrap_or(16);
+        sqlx::query_scalar("SELECT (max_claims + COALESCE((SELECT SUM(b.claim_chunks) FROM player_bonuses b JOIN guild_members m ON m.uuid = b.uuid WHERE m.guild_id = guilds.id), 0)) FROM guilds WHERE id = ?").bind(&guild_id).fetch_one(&state.db).await.unwrap_or(16);
 
     let current_claims: i64 =
         sqlx::query_scalar("SELECT COUNT(*) FROM guild_claims WHERE guild_id = ?").bind(&guild_id).fetch_one(&state.db).await.unwrap_or(0);
@@ -1250,7 +1250,7 @@ pub async fn server_get_player_guild(
     };
 
     let guild_opt: Option<(String, String, String, String, String, i64, i64, i64)> =
-        sqlx::query_as("SELECT name, tag, description, motd, leader_uuid, level, xp, max_claims FROM guilds WHERE id = ?")
+        sqlx::query_as("SELECT name, tag, description, motd, leader_uuid, level, xp, (max_claims + COALESCE((SELECT SUM(b.claim_chunks) FROM player_bonuses b JOIN guild_members m ON m.uuid = b.uuid WHERE m.guild_id = guilds.id), 0)) FROM guilds WHERE id = ?")
             .bind(&guild_id)
             .fetch_optional(&state.db)
             .await?;

panel/server/src/routes/leveling.rs

@@ -30,6 +30,7 @@ pub async fn grant_rewards(tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, uuid: &
         let inserted = sqlx::query("INSERT OR IGNORE INTO granted_rewards(uuid, reward_id, granted_at) VALUES (?, ?, ?)")
             .bind(uuid).bind(id).bind(now).execute(&mut **tx).await?.rows_affected();
         if inserted == 0 { continue; }
+        crate::rewards::enqueue(&mut **tx, uuid, "level_reward", &id.to_string()).await?;
         match kind.as_str() {
             "title" if scope == "global" => {
                 sqlx::query("UPDATE user_levels SET title = ? WHERE uuid = ?").bind(&name).bind(uuid).execute(&mut **tx).await?;
@@ -420,6 +421,7 @@ pub async fn admin_delete_reward(
     Path(id): Path<i64>,
     State(state): State<AppState>,
 ) -> AppResult<Json<Value>> {
+    sqlx::query("DELETE FROM reward_bundles WHERE source_type = 'level_reward' AND source_id = ?").bind(id.to_string()).execute(&state.db).await?;
     sqlx::query("DELETE FROM level_rewards WHERE id = ?")
         .bind(id)
         .execute(&state.db)

panel/server/src/routes/mod.rs

@@ -13,6 +13,7 @@ pub mod landing;
 pub mod worldmap;
 pub mod map;
 pub mod leveling;
+pub mod rewards;
 pub mod meta;
 pub mod progression;
 pub mod public;
@@ -74,6 +75,7 @@ pub fn api(state: &AppState) -> Router<AppState> {
         .route("/quests/my", get(quests::get_my_quests))
         .route("/quests/me", get(quests::get_my_quests))
         .route("/quests/{id}/claim", post(quests::claim_quest))
+        .route("/reward-bundles", get(rewards::public_summaries))
         // Achievements
         .route("/achievements/my", get(achievements::get_my_achievements))
         .route("/achievements/me", get(achievements::get_my_achievements))
@@ -173,6 +175,8 @@ pub fn api(state: &AppState) -> Router<AppState> {
         .route("/rewards/{id}", put(leveling::admin_update_reward).delete(leveling::admin_delete_reward))
         .route("/levels/rewards", get(leveling::list_rewards).post(leveling::admin_create_reward))
         .route("/levels/rewards/{id}", put(leveling::admin_update_reward).delete(leveling::admin_delete_reward))
+        .route("/reward-bundles/{kind}/{id}", get(rewards::admin_get).put(rewards::admin_put))
+        .route("/reward-deliveries", get(rewards::admin_deliveries))
         .route("/quests", get(quests::admin_list_quests).post(quests::admin_create_quest))
         .route("/quests/{id}", put(quests::admin_update_quest).delete(quests::admin_delete_quest))
         .route("/achievements", get(achievements::admin_list_achievements).post(achievements::admin_create_achievement))
@@ -208,6 +212,8 @@ pub fn api(state: &AppState) -> Router<AppState> {
         .route("/guilds/leave", post(guilds::server_guild_leave))
         .route("/integrations/report", post(integrations::server_report))
         .route("/actions/poll", post(map::poll_actions))
+        .route("/rewards/poll", post(rewards::game_poll))
+        .route("/rewards/ack", post(rewards::game_ack))
         .route("/guilds/invite/send", post(map::server_invite_send))
         .route("/guilds/invite/list", post(map::server_invites))
         .route("/guilds/invite/respond", post(map::server_invite_respond))

panel/server/src/routes/quests.rs

@@ -296,7 +296,9 @@ pub async fn claim_quest(
         .await?;
 
     crate::routes::leveling::grant_rewards(&mut tx, &auth.uuid, &now).await?;
+    crate::rewards::enqueue(&mut tx, &auth.uuid, "quest", &quest_id).await?;
     tx.commit().await?;
+    let _ = crate::rewards::process_queue(&state).await;
 
     Ok(Json(serde_json::json!({
         "ok": true,
@@ -480,6 +482,7 @@ pub async fn admin_delete_quest(
     Path(id): Path<String>,
     State(state): State<AppState>,
 ) -> AppResult<Json<Value>> {
+    sqlx::query("DELETE FROM reward_bundles WHERE source_type = 'quest' AND source_id = ?").bind(&id).execute(&state.db).await?;
     sqlx::query("DELETE FROM quests WHERE id = ?")
         .bind(&id)
         .execute(&state.db)

panel/server/src/routes/rewards.rs

@@ -0,0 +1,98 @@
+//! Admin editing of reward bundles, what players see of them, and the game server's delivery endpoints.
+
+use crate::auth::{AdminUser, AuthUser};
+use crate::error::{AppError, AppResult};
+use crate::rewards;
+use crate::routes::servers::GameServer;
+use crate::state::AppState;
+use axum::extract::{Path, Query, State};
+use axum::Json;
+use serde::Deserialize;
+use serde_json::{json, Value};
+use std::collections::BTreeMap;
+
+pub async fn admin_get(_: AdminUser, State(state): State<AppState>, Path((kind, id)): Path<(String, String)>) -> AppResult<Json<Value>> {
+    rewards::check_source(&kind)?;
+    let mut conn = state.db.acquire().await?;
+    Ok(Json(json!({ "actions": rewards::bundle(&mut conn, &kind, &id).await? })))
+}
+
+#[derive(Deserialize)]
+pub struct BundleBody {
+    actions: Value,
+}
+
+pub async fn admin_put(_: AdminUser, State(state): State<AppState>, Path((kind, id)): Path<(String, String)>, Json(body): Json<BundleBody>) -> AppResult<Json<Value>> {
+    rewards::check_source(&kind)?;
+    let actions = rewards::validate(&body.actions)?;
+    if actions.is_empty() {
+        sqlx::query("DELETE FROM reward_bundles WHERE source_type = ? AND source_id = ?").bind(&kind).bind(&id).execute(&state.db).await?;
+    } else {
+        sqlx::query("INSERT INTO reward_bundles (source_type, source_id, actions) VALUES (?, ?, ?) ON CONFLICT(source_type, source_id) DO UPDATE SET actions = excluded.actions")
+            .bind(&kind).bind(&id).bind(serde_json::to_string(&actions)?).execute(&state.db).await?;
+    }
+    Ok(Json(json!({ "ok": true, "actions": actions })))
+}
+
+#[derive(Deserialize)]
+pub struct LogQuery {
+    #[serde(default)]
+    limit: Option<i64>,
+}
+
+/// What was handed out (or is waiting, or failed), newest first.
+pub async fn admin_deliveries(_: AdminUser, State(state): State<AppState>, Query(q): Query<LogQuery>) -> AppResult<Json<Value>> {
+    let rows: Vec<(i64, String, String, Option<i64>, String, String, String, Option<String>, i64, Option<String>)> = sqlx::query_as(
+        "SELECT d.id, COALESCE(u.username, d.uuid), d.kind, d.server_id, d.payload, d.source, d.created_at, d.delivered_at, d.attempts, d.error
+         FROM reward_deliveries d LEFT JOIN users u ON u.uuid = d.uuid ORDER BY d.id DESC LIMIT ?",
+    )
+    .bind(q.limit.unwrap_or(100).clamp(1, 500))
+    .fetch_all(&state.db)
+    .await?;
+    Ok(Json(json!(rows.into_iter().map(|(id, player, kind, server, payload, source, at, delivered, attempts, error)| {
+        let p: Value = serde_json::from_str(&payload).unwrap_or(json!({}));
+        json!({ "id": id, "player": player, "kind": kind, "server_id": server, "summary": rewards::describe(&p), "source": source, "created_at": at,
+                "delivered_at": delivered, "attempts": attempts, "error": error })
+    }).collect::<Vec<_>>())))
+}
+
+/// `{"quest:daily_miner": ["$200", "Diamond x3"], ...}` for the launcher to show next to each reward.
+pub async fn public_summaries(_: AuthUser, State(state): State<AppState>) -> AppResult<Json<Value>> {
+    let rows: Vec<(String, String, String)> = sqlx::query_as("SELECT source_type, source_id, actions FROM reward_bundles").fetch_all(&state.db).await?;
+    let mut out = BTreeMap::new();
+    for (kind, id, actions) in rows {
+        let list: Vec<Value> = serde_json::from_str(&actions).unwrap_or_default();
+        if !list.is_empty() {
+            out.insert(format!("{kind}:{id}"), list.iter().map(rewards::describe).collect::<Vec<_>>());
+        }
+    }
+    Ok(Json(json!(out)))
+}
+
+pub async fn game_poll(GameServer(server): GameServer, State(state): State<AppState>) -> AppResult<Json<Value>> {
+    Ok(Json(json!({ "deliveries": rewards::poll(&state, server.id).await? })))
+}
+
+#[derive(Deserialize)]
+pub struct AckBody {
+    #[serde(default)]
+    done: Vec<i64>,
+    #[serde(default)]
+    failed: Vec<Failure>,
+}
+
+#[derive(Deserialize)]
+pub struct Failure {
+    id: i64,
+    #[serde(default)]
+    error: String,
+}
+
+pub async fn game_ack(GameServer(_): GameServer, State(state): State<AppState>, Json(body): Json<AckBody>) -> AppResult<Json<Value>> {
+    if body.done.len() + body.failed.len() > 200 {
+        return Err(AppError::bad_request("too many ids"));
+    }
+    let failed: Vec<(i64, String)> = body.failed.into_iter().map(|f| (f.id, f.error)).collect();
+    rewards::ack(&state, &body.done, &failed).await?;
+    Ok(Json(json!({ "ok": true })))
+}

panel/server/tests/rewards.rs

@@ -0,0 +1,118 @@
+//! Bundles of rewards beyond XP: set by an admin, applied when earned, and delivered to the game server.
+
+mod common;
+use common::*;
+
+#[tokio::test]
+async fn an_achievement_can_pay_money_chunks_items_and_permissions() {
+    let t = setup().await;
+    let admin = t.login("admin", "supersecret").await;
+    t.call("POST", "/api/admin/users", Some(&admin), Some(json!({"username": "Steve", "password": "password123"}))).await;
+    let steve_token = t.login("Steve", "password123").await;
+    let steve = t.uuid("Steve").await;
+    let (_, srv) = t.call("POST", "/api/admin/servers", Some(&admin), Some(json!({"name": "SMP", "instance_id": "smp"}))).await;
+    let sid = srv["server"]["id"].as_i64().unwrap();
+    let key = srv["token"].as_str().unwrap().to_string();
+    let game = |path: &str, body: Value| {
+        let (t, key, path) = (&t, key.clone(), path.to_string());
+        async move { t.call("POST", &format!("/api/server/v1/{path}"), Some(&key), Some(body)).await }
+    };
+    let (s, ach) = t.call("POST", "/api/admin/achievements", Some(&admin), Some(json!({
+        "id": "ach_test", "title": "Tester", "description": "x", "category": "general", "icon_frame": "task", "icon_item": "minecraft:stone",
+        "icon_bg": "#000000", "icon_border": "#ffffff", "xp_reward": 10}))).await;
+    assert_eq!(s, StatusCode::OK, "{ach}");
+
+    // Bad bundles are refused; good ones are normalised.
+    for a in [json!({"type": "money", "amount": -5}), json!({"type": "item", "item": "bad item!", "amount": 1}), json!({"type": "nope"})] {
+        assert_eq!(t.call("PUT", "/api/admin/reward-bundles/achievement/ach_test", Some(&admin), Some(json!({"actions": [a]}))).await.0, StatusCode::BAD_REQUEST);
+    }
+    assert_eq!(t.call("PUT", "/api/admin/reward-bundles/banana/x", Some(&admin), Some(json!({"actions": []}))).await.0, StatusCode::BAD_REQUEST);
+    let (s, saved) = t.call("PUT", "/api/admin/reward-bundles/achievement/ach_test", Some(&admin), Some(json!({"actions": [
+        {"type": "money", "amount": 250},
+        {"type": "claim_chunks", "amount": 8},
+        {"type": "item", "item": "diamond", "amount": 3},
+        {"type": "permission", "node": "essentials.fly", "minutes": 10080},
+        {"type": "group", "group": "vip"},
+        {"type": "message", "text": "Well done!"}]}))).await;
+    assert_eq!(s, StatusCode::OK, "{saved}");
+    assert_eq!(saved["actions"][2]["item"], "minecraft:diamond");
+
+    // Players see short summaries (never command text).
+    let (_, sums) = t.call("GET", "/api/v1/reward-bundles", Some(&steve_token), None).await;
+    assert_eq!(sums["achievement:ach_test"][0], "$250.00");
+    assert_eq!(sums["achievement:ach_test"][2], "Diamond x3");
+    assert_eq!(sums["achievement:ach_test"][3], "Permission: essentials.fly for 7 days");
+
+    // Earning it applies the panel-side rewards straight away…
+    sqlx::query("INSERT INTO server_online (server_id, uuid, name, joined_at) VALUES (?, ?, 'Steve', 'now')").bind(sid).bind(&steve).execute(&t.db).await.unwrap();
+    sqlx::query("INSERT INTO user_achievements (user_uuid, achievement_id, unlocked_at) VALUES (?, 'ach_test', 'now')").bind(&steve).execute(&t.db).await.unwrap();
+    let (s, polled) = game("rewards/poll", json!({})).await;
+    assert_eq!(s, StatusCode::OK, "{polled}");
+    let kinds: Vec<String> = polled["deliveries"].as_array().unwrap().iter().map(|d| d["kind"].as_str().unwrap().to_string()).collect();
+    assert_eq!(kinds, ["item", "permission", "group", "message"], "{polled}");
+    assert_eq!(polled["deliveries"][0]["payload"]["item"], "minecraft:diamond");
+    assert_eq!(polled["deliveries"][0]["name"], "Steve");
+    let balance: f64 = sqlx::query_scalar("SELECT balance FROM server_economy WHERE server_id = ? AND uuid = ?").bind(sid).bind(&steve).fetch_one(&t.db).await.unwrap();
+    assert_eq!(balance, 1250.0, "the starting 1000 plus 250");
+    let chunks: i64 = sqlx::query_scalar("SELECT claim_chunks FROM player_bonuses WHERE uuid = ?").bind(&steve).fetch_one(&t.db).await.unwrap();
+    assert_eq!(chunks, 8);
+
+    // …and a delivery is handed out once until acknowledged.
+    assert!(game("rewards/poll", json!({})).await.1["deliveries"].as_array().unwrap().is_empty());
+    let ids: Vec<i64> = polled["deliveries"].as_array().unwrap().iter().map(|d| d["id"].as_i64().unwrap()).collect();
+    let (s, _) = game("rewards/ack", json!({"done": [ids[0], ids[1]], "failed": [{"id": ids[2], "error": "no such group"}]})).await;
+    assert_eq!(s, StatusCode::OK);
+    let (_, log) = t.call("GET", "/api/admin/reward-deliveries", Some(&admin), None).await;
+    let log = log.as_array().unwrap();
+    assert_eq!(log.len(), 4);
+    assert!(log.iter().any(|d| d["error"] == "no such group"));
+    assert_eq!(log.iter().filter(|d| !d["delivered_at"].is_null()).count(), 2);
+
+    // The extra chunks lift the player's guild's claim limit.
+    let (_, g) = t.call("POST", "/api/v1/guilds", Some(&steve_token), Some(json!({"instance_id": "smp", "name": "Iron", "tag": "IRON"}))).await;
+    let (_, detail) = t.call("GET", &format!("/api/v1/guilds/{}", g["id"].as_str().unwrap()), Some(&steve_token), None).await;
+    assert_eq!(detail["max_claims"].as_i64(), Some(16 + 8), "{detail}");
+}
+
+#[tokio::test]
+async fn deliveries_wait_for_the_player_to_be_online() {
+    let t = setup().await;
+    let admin = t.login("admin", "supersecret").await;
+    t.call("POST", "/api/admin/users", Some(&admin), Some(json!({"username": "Alex", "password": "password123"}))).await;
+    let alex = t.uuid("Alex").await;
+    let (_, srv) = t.call("POST", "/api/admin/servers", Some(&admin), Some(json!({"name": "SMP", "instance_id": "smp"}))).await;
+    let sid = srv["server"]["id"].as_i64().unwrap();
+    let key = srv["token"].as_str().unwrap().to_string();
+    sqlx::query("INSERT INTO reward_deliveries (uuid, kind, payload, source, created_at) VALUES (?, 'item', '{\"type\":\"item\",\"item\":\"minecraft:apple\",\"amount\":1}', 'x', 'now')").bind(&alex).execute(&t.db).await.unwrap();
+    let poll = || async { t.call("POST", "/api/server/v1/rewards/poll", Some(&key), Some(json!({}))).await.1 };
+    assert!(poll().await["deliveries"].as_array().unwrap().is_empty(), "offline players get nothing yet");
+    sqlx::query("INSERT INTO server_online (server_id, uuid, name, joined_at) VALUES (?, ?, 'Alex', 'now')").bind(sid).bind(&alex).execute(&t.db).await.unwrap();
+    assert_eq!(poll().await["deliveries"].as_array().unwrap().len(), 1);
+}
+
+#[tokio::test]
+async fn a_rank_milestone_can_grant_more_than_its_title() {
+    let t = setup().await;
+    let admin = t.login("admin", "supersecret").await;
+    t.call("POST", "/api/admin/users", Some(&admin), Some(json!({"username": "Alex", "password": "password123"}))).await;
+    let alex = t.uuid("Alex").await;
+    let (_, srv) = t.call("POST", "/api/admin/servers", Some(&admin), Some(json!({"name": "SMP", "instance_id": "smp"}))).await;
+    let sid = srv["server"]["id"].as_i64().unwrap();
+    let key = srv["token"].as_str().unwrap().to_string();
+    let (_, made) = t.call("POST", "/api/admin/rewards", Some(&admin), Some(json!({"level_type": "global", "level_req": 3, "reward_type": "title", "reward_name": "Veteran"}))).await;
+    let id = made["id"].as_i64().unwrap();
+    let (s, _) = t.call("PUT", &format!("/api/admin/reward-bundles/level_reward/{id}"), Some(&admin), Some(json!({"actions": [
+        {"type": "money", "amount": 500}, {"type": "command", "command": "/give {player} cake 1"}]}))).await;
+    assert_eq!(s, StatusCode::OK);
+    sqlx::query("INSERT INTO server_online (server_id, uuid, name, joined_at) VALUES (?, ?, 'Alex', 'now')").bind(sid).bind(&alex).execute(&t.db).await.unwrap();
+    let (s, r) = t.call("POST", "/api/server/v1/player/xp", Some(&key), Some(json!({"operation_id": "lv", "uuid": alex, "scope": "global", "amount": 100000}))).await;
+    assert_eq!(s, StatusCode::OK, "{r}");
+    let (_, polled) = t.call("POST", "/api/server/v1/rewards/poll", Some(&key), Some(json!({}))).await;
+    assert_eq!(polled["deliveries"][0]["kind"], "command", "{polled}");
+    assert_eq!(polled["deliveries"][0]["payload"]["command"], "give {player} cake 1");
+    let balance: f64 = sqlx::query_scalar("SELECT balance FROM server_economy WHERE server_id = ? AND uuid = ?").bind(sid).bind(&alex).fetch_one(&t.db).await.unwrap();
+    assert_eq!(balance, 1500.0);
+    // Earning the same milestone twice pays once.
+    t.call("POST", "/api/server/v1/player/xp", Some(&key), Some(json!({"operation_id": "lv2", "uuid": alex, "scope": "global", "amount": 100}))).await;
+    assert!(t.call("POST", "/api/server/v1/rewards/poll", Some(&key), Some(json!({}))).await.1["deliveries"].as_array().unwrap().is_empty());
+}