mirror of
https://git.ethanthesleepy.one/ethanaobrien/ew
synced 2026-08-26 15:02:18 +08:00
Compare commits
3 Commits
75613dafb0
...
0df2741251
| Author | SHA1 | Date | |
|---|---|---|---|
|
0df2741251 | ||
|
|
cf05ffe8fb | ||
|
|
27d7b8de82 |
5
.gitignore
vendored
5
.gitignore
vendored
@@ -12,3 +12,8 @@ ndk/
|
||||
.DS_Store
|
||||
custom_songs/
|
||||
custom_cards/
|
||||
|
||||
# local-only trees — never commit (35GB between them)
|
||||
/android/
|
||||
/assets.bak/
|
||||
/assets.old/
|
||||
|
||||
19
Cargo.lock
generated
19
Cargo.lock
generated
@@ -223,6 +223,22 @@ dependencies = [
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "actix-ws"
|
||||
version = "0.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "decf53c3cdd63dd6f289980b430238f9a2f6d19f8bce8e418272e08d3da43f0f"
|
||||
dependencies = [
|
||||
"actix-codec",
|
||||
"actix-http",
|
||||
"actix-web",
|
||||
"bytestring",
|
||||
"futures-core",
|
||||
"futures-sink",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "adler2"
|
||||
version = "2.0.1"
|
||||
@@ -1073,6 +1089,7 @@ version = "1.1.0"
|
||||
dependencies = [
|
||||
"actix-multipart",
|
||||
"actix-web",
|
||||
"actix-ws",
|
||||
"aes",
|
||||
"argon2",
|
||||
"base64",
|
||||
@@ -1103,6 +1120,7 @@ dependencies = [
|
||||
"sha1",
|
||||
"sha2",
|
||||
"symphonia",
|
||||
"tokio",
|
||||
"ureq",
|
||||
"urlencoding",
|
||||
"uuid",
|
||||
@@ -3209,6 +3227,7 @@ dependencies = [
|
||||
"bytes",
|
||||
"futures-core",
|
||||
"futures-sink",
|
||||
"futures-util",
|
||||
"pin-project-lite",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
@@ -32,6 +32,10 @@ include_dir = "0.7.4"
|
||||
jzon = "0.12.5"
|
||||
csv = "1.3"
|
||||
actix-multipart = "0.8"
|
||||
# Multi-live relay: WebSockets over actix-web 4 without the actor runtime
|
||||
actix-ws = "0.4"
|
||||
# tokio is already in the tree under actix-rt; only the mpsc channels are used
|
||||
tokio = { version = "1", features = ["sync"] }
|
||||
futures-util = "0.3"
|
||||
image = "0.25"
|
||||
zip = { version = "8.6", default-features = false, features = ["deflate"] }
|
||||
|
||||
@@ -37,6 +37,10 @@ pub async fn run_server(in_thread: bool) -> std::io::Result<()> {
|
||||
|
||||
router::custom_song::migrate::run();
|
||||
router::custom_song::sweep_audio();
|
||||
// The multi-live relay's expiry timers, on the system arbiter rather than on whichever
|
||||
// HTTP worker happened to serve the first WebSocket upgrade — a worker panic must not
|
||||
// be able to strand every room's seats for the life of the process.
|
||||
router::multi_live::start_sweeper();
|
||||
|
||||
let rv = HttpServer::new(|| App::new()
|
||||
//.wrap(Cors::permissive())
|
||||
|
||||
@@ -10,6 +10,7 @@ pub mod home;
|
||||
pub mod lottery;
|
||||
pub mod friend;
|
||||
pub mod live;
|
||||
pub mod multi_live;
|
||||
pub mod event;
|
||||
pub mod chat;
|
||||
pub mod story;
|
||||
@@ -274,6 +275,7 @@ pub fn configure(cfg: &mut actix_web::web::ServiceConfig) {
|
||||
.configure(login::routes)
|
||||
.configure(lottery::routes)
|
||||
.configure(mission::routes)
|
||||
.configure(multi_live::routes)
|
||||
.configure(notice::routes)
|
||||
.configure(purchase::routes)
|
||||
.configure(serial_code::routes)
|
||||
|
||||
@@ -7,11 +7,33 @@ pub fn routes(cfg: &mut web::ServiceConfig) {
|
||||
cfg.service(
|
||||
web::scope("/chat")
|
||||
.route("/home", web::post().to(home))
|
||||
.route("/talk/get_stamp", web::get().to(get_stamp))
|
||||
.route("/talk/start", web::post().to(start))
|
||||
.route("/talk/end", web::post().to(end))
|
||||
);
|
||||
}
|
||||
|
||||
// The stamps this account owns. ew does not track stamp unlocks, so everyone has the
|
||||
// masterdata initial set — the same list /chat/home reports, deliberately from one
|
||||
// source so the two endpoints can never disagree.
|
||||
fn owned_stamp_ids() -> JsonValue {
|
||||
databases::INITIAL_CHAT_STAMPS.clone()
|
||||
}
|
||||
|
||||
// GET /api/chat/talk/get_stamp (Protocol.send_get_stamp, FuncId.GET_STAMP).
|
||||
// RecvGetStampRData carries a single `master_chat_stamp_ids` array, which its Notify()
|
||||
// feeds to CallOnUpdateChatStampListNotify — the same sink RecvChatHomeRData uses, so
|
||||
// the stamp picker ends up with whatever this returns. The client sends no parameters.
|
||||
//
|
||||
// This endpoint appears in neither official capture (0 hits across the 288MB JP and
|
||||
// 1.4GB EN logs), so the shape is taken from the client class and the contents from the
|
||||
// /chat/home captures that do exist and carry the same field.
|
||||
async fn get_stamp(Login(_key): Login) -> impl Responder {
|
||||
Api(Some(object!{
|
||||
"master_chat_stamp_ids": owned_stamp_ids()
|
||||
}))
|
||||
}
|
||||
|
||||
pub fn add_chat(id: i64, num: i64, chats: &mut JsonValue) -> bool {
|
||||
for data in chats.members() {
|
||||
if data["chat_id"] == id && data["room_id"] == num {
|
||||
@@ -48,7 +70,7 @@ async fn home(Login(key): Login) -> impl Responder {
|
||||
Api(Some(object!{
|
||||
"progress_list": chats,
|
||||
"master_chat_room_ids": rooms,
|
||||
"master_chat_stamp_ids": [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,43,44,45,46,48,49,50,51,52,53,54,55,56,57,58,59,60,61,62,63,64,65,66,67,68,69,70,71,72,73,74,75,76,77,78,79,80,81,82,83,84,85,86,87,88,89,90,91,92,93,94,95,96,11001003,22001001,33001001,44001002],
|
||||
"master_chat_stamp_ids": owned_stamp_ids(),
|
||||
"master_chat_attachment_ids": []
|
||||
}))
|
||||
}
|
||||
@@ -75,3 +97,41 @@ async fn end(Session { key, body }: Session) -> impl Responder {
|
||||
|
||||
Api(Some(array![]))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
// Verbatim from an official /api/chat/home capture (JP log, the 65-occurrence
|
||||
// baseline seen on accounts that had earned no extra stamps yet). Both the JP and EN
|
||||
// chat_stamp tables reproduce it exactly from _initialStamp, which is what lets
|
||||
// get_stamp serve masterdata instead of a hardcoded literal.
|
||||
const OFFICIAL_INITIAL_STAMPS: [i64; 97] = [
|
||||
1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,19,20,21,22,23,24,25,26,27,28,29,30,
|
||||
31,32,33,34,35,36,37,38,39,40,41,43,44,45,46,48,49,50,51,52,53,54,55,56,57,58,
|
||||
59,60,61,62,63,64,65,66,67,68,69,70,71,72,73,74,75,76,77,78,79,80,81,82,83,84,
|
||||
85,86,87,88,89,90,91,92,93,94,95,96,11001003,22001001,33001001,44001002
|
||||
];
|
||||
|
||||
#[test]
|
||||
fn the_initial_stamp_set_matches_official() {
|
||||
let ids: Vec<i64> = owned_stamp_ids().members().map(|s| s.as_i64().unwrap()).collect();
|
||||
assert_eq!(ids, OFFICIAL_INITIAL_STAMPS.to_vec());
|
||||
// Order matters: the official list is masterdata order, not sorted (the 11xxxxxx
|
||||
// band trails the small ids).
|
||||
assert_eq!(ids.first(), Some(&1));
|
||||
assert_eq!(ids.last(), Some(&44001002));
|
||||
// The gaps are real - 18, 42 and 47 are not initial stamps.
|
||||
for missing in [18, 42, 47] {
|
||||
assert!(!ids.contains(&missing), "{missing} should not be an initial stamp");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn get_stamp_and_chat_home_cannot_disagree() {
|
||||
// Both endpoints read the one source; this is what stops /chat/home's list and
|
||||
// the stamp picker's list from drifting apart.
|
||||
assert_eq!(owned_stamp_ids(), *databases::INITIAL_CHAT_STAMPS);
|
||||
assert!(!owned_stamp_ids().is_empty());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -68,14 +68,14 @@ fn setup_tables(conn: &rusqlite::Connection) {
|
||||
);").unwrap();
|
||||
}
|
||||
|
||||
fn update_live_score(id: i64, uid: i64, score: i64) {
|
||||
if uid == 0 || score == 0 {
|
||||
return;
|
||||
}
|
||||
|
||||
let info = DATABASE.lock_and_select("SELECT score_data FROM scores WHERE live_id=?1", params!(id)).unwrap_or(String::from("[]"));
|
||||
let scores = jzon::parse(&info).unwrap();
|
||||
|
||||
// Merges this play into the song's top-10 board. Pure so the whole read-modify-write can
|
||||
// sit inside one transaction, and so the keep-best rule is testable on its own.
|
||||
//
|
||||
// The board is per-song, not per-account: `scores` is keyed by live_id alone and the user
|
||||
// lives inside the JSON blob. A user already on the board keeps whichever of their two
|
||||
// scores is higher — a replay that does not beat the stored one is dropped (`None`), which
|
||||
// is what makes a repeated or duplicated end idempotent rather than additive.
|
||||
fn merge_live_score(scores: &JsonValue, uid: i64, score: i64) -> Option<JsonValue> {
|
||||
let mut result = array![];
|
||||
let mut current = 0;
|
||||
let mut added = false;
|
||||
@@ -99,7 +99,8 @@ fn update_live_score(id: i64, uid: i64, score: i64) {
|
||||
}
|
||||
}
|
||||
if scores[i]["user"].as_i64().unwrap() == uid && !added {
|
||||
return;
|
||||
// Already on the board with a better score — keep it, drop this play.
|
||||
return None;
|
||||
}
|
||||
if scores[i]["user"].as_i64().unwrap() == uid {
|
||||
continue;
|
||||
@@ -108,12 +109,40 @@ fn update_live_score(id: i64, uid: i64, score: i64) {
|
||||
current += 1;
|
||||
}
|
||||
|
||||
if added {
|
||||
if DATABASE.lock_and_select("SELECT live_id FROM scores WHERE live_id=?1", params!(id)).is_ok() {
|
||||
DATABASE.lock_and_exec("UPDATE scores SET score_data=?1 WHERE live_id=?2", params!(jzon::stringify(result), id));
|
||||
} else {
|
||||
DATABASE.lock_and_exec("INSERT INTO scores (score_data, live_id) VALUES (?1, ?2)", params!(jzon::stringify(result), id));
|
||||
if added { Some(result) } else { None }
|
||||
}
|
||||
|
||||
fn update_live_score(id: i64, uid: i64, score: i64) {
|
||||
if uid == 0 || score == 0 {
|
||||
return;
|
||||
}
|
||||
|
||||
// One transaction for read + merge + write. Previously this SELECTed to choose between
|
||||
// UPDATE and INSERT on separate connections, so two ends landing together for a song
|
||||
// with no row yet both took the INSERT branch and the loser panicked the worker on
|
||||
// `UNIQUE constraint failed: scores.live_id`. Two clients finishing the same multi
|
||||
// live make that the normal case, not a rare race.
|
||||
let write = DATABASE.lock_and_transact(|conn| {
|
||||
let stored: String = conn
|
||||
.query_row("SELECT score_data FROM scores WHERE live_id=?1", params!(id), |row| row.get(0))
|
||||
.unwrap_or_else(|_| String::from("[]"));
|
||||
let scores = jzon::parse(&stored).unwrap_or_else(|_| array![]);
|
||||
|
||||
let Some(result) = merge_live_score(&scores, uid, score) else {
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
// Atomic upsert: no branch left for a concurrent writer to slip between.
|
||||
conn.execute(
|
||||
"INSERT INTO scores (live_id, score_data) VALUES (?1, ?2)
|
||||
ON CONFLICT(live_id) DO UPDATE SET score_data=excluded.score_data",
|
||||
params!(id, jzon::stringify(result))
|
||||
)?;
|
||||
Ok(())
|
||||
});
|
||||
|
||||
if let Err(e) = write {
|
||||
println!("Failed to record score for live {id}: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -129,27 +158,44 @@ pub fn invalidate_cache() {
|
||||
crate::lock_onto_mutex!(CACHED_HTML_DATA).take();
|
||||
}
|
||||
|
||||
// The clear-rate counter column this play lands in, or None for a level outside 1-4.
|
||||
// Names come from this closed set, never from request data, so it is safe to interpolate.
|
||||
fn clear_rate_column(level: i32, failed: bool) -> Option<&'static str> {
|
||||
let tier = match level {
|
||||
1 => "normal",
|
||||
2 => "hard",
|
||||
3 => "expert",
|
||||
4 => "master",
|
||||
_ => return None
|
||||
};
|
||||
Some(match (tier, failed) {
|
||||
("normal", true) => "normal_failed", ("normal", false) => "normal_pass",
|
||||
("hard", true) => "hard_failed", ("hard", false) => "hard_pass",
|
||||
("expert", true) => "expert_failed", ("expert", false) => "expert_pass",
|
||||
(_, true) => "master_failed", (_, false) => "master_pass"
|
||||
})
|
||||
}
|
||||
|
||||
pub fn live_completed(id: i64, level: i32, failed: bool, score: i64, uid: i64) {
|
||||
update_live_score(id, uid, score);
|
||||
match DATABASE.get_live_data(id) {
|
||||
Ok(info) => {
|
||||
let value = format!("{}_{}",
|
||||
if 1 == level { "normal" } else if 2 == level { "hard" } else if 3 == level { "expert" } else { "master" },
|
||||
if failed { "failed" } else { "pass" }
|
||||
);
|
||||
let new_info = if 1 == level && failed { info.normal_failed }
|
||||
else if 1 == level && !failed { info.normal_pass }
|
||||
else if 2 == level && failed { info.hard_failed }
|
||||
else if 2 == level && !failed { info.hard_pass }
|
||||
else if 3 == level && failed { info.expert_failed }
|
||||
else if 3 == level && !failed { info.expert_pass }
|
||||
else if 4 == level && failed { info.master_failed }
|
||||
else if 4 == level && !failed { info.master_pass } else { return; };
|
||||
|
||||
DATABASE.lock_and_exec(&format!("UPDATE lives SET {}=?1 WHERE live_id=?2", value), params!(new_info + 1, info.live_id));
|
||||
},
|
||||
Err(_) => {
|
||||
DATABASE.lock_and_exec("INSERT INTO lives (live_id, normal_failed, normal_pass, hard_failed, hard_pass, expert_failed, expert_pass, master_failed, master_pass) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", params!(
|
||||
let Some(column) = clear_rate_column(level, failed) else {
|
||||
return;
|
||||
};
|
||||
|
||||
// `lives` is keyed by live_id alone and had the same select-then-INSERT-or-UPDATE
|
||||
// split as `scores`, so it could panic the same way on a first-ever concurrent play
|
||||
// (and lost counts whenever two plays overlapped). One upsert does both branches
|
||||
// atomically and increments in SQL rather than read-modify-write in Rust.
|
||||
let write = DATABASE.lock_and_transact(|conn| {
|
||||
conn.execute(
|
||||
&format!(
|
||||
"INSERT INTO lives (live_id, normal_failed, normal_pass, hard_failed, hard_pass,
|
||||
expert_failed, expert_pass, master_failed, master_pass)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
|
||||
ON CONFLICT(live_id) DO UPDATE SET {column} = {column} + 1"
|
||||
),
|
||||
params!(
|
||||
id,
|
||||
if 1 == level && failed { 1 } else { 0 },
|
||||
if 1 == level && !failed { 1 } else { 0 },
|
||||
@@ -159,9 +205,14 @@ pub fn live_completed(id: i64, level: i32, failed: bool, score: i64, uid: i64) {
|
||||
if 3 == level && !failed { 1 } else { 0 },
|
||||
if 4 == level && failed { 1 } else { 0 },
|
||||
if 4 == level && !failed { 1 } else { 0 }
|
||||
));
|
||||
},
|
||||
};
|
||||
)
|
||||
)?;
|
||||
Ok(())
|
||||
});
|
||||
|
||||
if let Err(e) = write {
|
||||
println!("Failed to record clear rate for live {id}: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
fn get_song_title(live_id: i32, english: bool) -> String {
|
||||
@@ -407,3 +458,107 @@ pub async fn clearrate_html(_req: HttpRequest) -> HttpResponse {
|
||||
.content_type(ContentType::html())
|
||||
.body(html)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn board(live_id: i64) -> JsonValue {
|
||||
let stored = DATABASE
|
||||
.lock_and_select("SELECT score_data FROM scores WHERE live_id=?1", params!(live_id))
|
||||
.unwrap_or_else(|_| String::from("[]"));
|
||||
jzon::parse(&stored).unwrap()
|
||||
}
|
||||
|
||||
fn passes(live_id: i64) -> i64 {
|
||||
DATABASE.get_live_data(live_id).map(|l| l.master_pass).unwrap_or(0)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn merge_keeps_the_users_best_score() {
|
||||
let existing = array![object!{user: 7, score: 900}];
|
||||
|
||||
// A better score replaces the stored one rather than adding a second entry.
|
||||
let better = merge_live_score(&existing, 7, 1000).expect("a better score is recorded");
|
||||
assert_eq!(better.len(), 1);
|
||||
assert_eq!(better[0]["score"].as_i64(), Some(1000));
|
||||
|
||||
// A worse or equal replay is dropped entirely — this is what makes a repeated
|
||||
// end idempotent instead of appending the same user twice.
|
||||
assert!(merge_live_score(&existing, 7, 800).is_none());
|
||||
assert!(merge_live_score(&existing, 7, 900).is_none());
|
||||
|
||||
// A different user is ranked against the board, best first.
|
||||
let other = merge_live_score(&existing, 8, 950).expect("a new user is recorded");
|
||||
assert_eq!(other.len(), 2);
|
||||
assert_eq!(other[0]["user"].as_i64(), Some(8));
|
||||
assert_eq!(other[1]["user"].as_i64(), Some(7));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_duplicate_end_does_not_double_the_board_entry() {
|
||||
let _lock = crate::runtime::lock_test_data_path();
|
||||
let live_id = 990001;
|
||||
|
||||
live_completed(live_id, 4, false, 500000, 4242);
|
||||
assert_eq!(board(live_id).len(), 1);
|
||||
assert_eq!(passes(live_id), 1);
|
||||
|
||||
// The second end for the same session: same user, same score. The board must not
|
||||
// grow, and this must not raise UNIQUE constraint failed: scores.live_id.
|
||||
live_completed(live_id, 4, false, 500000, 4242);
|
||||
assert_eq!(board(live_id).len(), 1, "the same user must not appear twice");
|
||||
assert_eq!(board(live_id)[0]["score"].as_i64(), Some(500000));
|
||||
}
|
||||
|
||||
// The actual regression: two ends for a song with no row yet, landing together. Both
|
||||
// used to take the INSERT branch and the loser unwrapped a ConstraintViolation into a
|
||||
// worker panic. Two clients finishing one multi live makes this the normal case.
|
||||
#[test]
|
||||
fn concurrent_first_plays_of_one_song_do_not_collide() {
|
||||
let _lock = crate::runtime::lock_test_data_path();
|
||||
let live_id = 990002;
|
||||
|
||||
std::thread::scope(|s| {
|
||||
for uid in [101i64, 102, 103, 104, 105, 106, 107, 108] {
|
||||
s.spawn(move || live_completed(live_id, 4, false, 400000 + uid, uid));
|
||||
}
|
||||
});
|
||||
|
||||
// Every writer landed: no row lost to a race, none lost to a swallowed error.
|
||||
assert_eq!(board(live_id).len(), 8);
|
||||
assert_eq!(passes(live_id), 8, "clear-rate counts must not be lost either");
|
||||
}
|
||||
|
||||
// The other half of /multi_live/end's score-board branch (the account's own high score
|
||||
// is pinned in live.rs): a public multi live reaches live_completed with uid 0, so the
|
||||
// play is counted and the board is left alone. /live/retire has always used the same
|
||||
// signal for a failed live.
|
||||
#[test]
|
||||
fn a_play_with_no_user_counts_the_clear_but_not_the_board() {
|
||||
let _lock = crate::runtime::lock_test_data_path();
|
||||
let live_id = 990003;
|
||||
|
||||
live_completed(live_id, 4, false, 500000, 0);
|
||||
assert_eq!(board(live_id).len(), 0, "an unranked play must not reach the board");
|
||||
assert_eq!(passes(live_id), 1, "but it is still a play of the song");
|
||||
|
||||
// The same score from a real account does land, which is the private-room path.
|
||||
live_completed(live_id, 4, false, 500000, 4243);
|
||||
assert_eq!(board(live_id).len(), 1);
|
||||
assert_eq!(passes(live_id), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn clear_rate_columns_cover_every_level() {
|
||||
assert_eq!(clear_rate_column(1, false), Some("normal_pass"));
|
||||
assert_eq!(clear_rate_column(1, true), Some("normal_failed"));
|
||||
assert_eq!(clear_rate_column(2, false), Some("hard_pass"));
|
||||
assert_eq!(clear_rate_column(3, true), Some("expert_failed"));
|
||||
assert_eq!(clear_rate_column(4, false), Some("master_pass"));
|
||||
assert_eq!(clear_rate_column(4, true), Some("master_failed"));
|
||||
// Level 0 (a skip ticket's "any level") writes no counter, as before.
|
||||
assert_eq!(clear_rate_column(0, false), None);
|
||||
assert_eq!(clear_rate_column(5, false), None);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -302,6 +302,57 @@ lazy_static! {
|
||||
info
|
||||
};
|
||||
|
||||
// const.csv keyed by _id. Values are strings in masterdata, exactly as the
|
||||
// client reads them (ConstMst._value + StringExtensions.ToIntOrDefault).
|
||||
pub static ref CONST: JsonValue = index_by(&t("const"), "id");
|
||||
|
||||
pub static ref LIVE_BOOST: JsonValue = index_by(&t("live_boost"), "value");
|
||||
|
||||
// The stamps every account starts with (chat_stamp._initialStamp), in masterdata
|
||||
// order. Officially this is the whole of a fresh account's master_chat_stamp_ids —
|
||||
// captured /api/chat/home responses open with exactly this list before an account's
|
||||
// earned stamps are appended (see chat::tests::the_initial_stamp_set_matches_official).
|
||||
pub static ref INITIAL_CHAT_STAMPS: JsonValue = {
|
||||
let mut ids = array![];
|
||||
for data in t("chat_stamp").members() {
|
||||
if data["initialStamp"].as_i64().unwrap_or(0) == 1 {
|
||||
ids.push(data["id"].clone()).unwrap();
|
||||
}
|
||||
}
|
||||
ids
|
||||
};
|
||||
|
||||
pub static ref EVENTS: JsonValue = index_by(&t("event"), "id");
|
||||
|
||||
// release_label.csv keyed by _id — the open/close window masterdata rows are gated
|
||||
// on. _openedAt / _closedAt are blank for the evergreen label (id 1).
|
||||
pub static ref RELEASE_LABEL: JsonValue = index_by(&t("release_label"), "id");
|
||||
|
||||
// event_score.csv keyed by _masterEventId (Shock.EventScoreMst) — the per-event
|
||||
// event-point yield of one live. Ratios are 1/10000, like every other ratio the
|
||||
// client divides by COMMON_CONST.RATIO_DIVISOR.
|
||||
pub static ref EVENT_SCORE: JsonValue = index_by(&t("event_score"), "masterEventId");
|
||||
|
||||
// music_level rows keyed "{masterMusicId}_{level}" — _fullCombo is the note
|
||||
// count the multi-live miss/great-perfect ratios are measured against.
|
||||
pub static ref MUSIC_LEVEL: JsonValue = {
|
||||
let mut info = object! {};
|
||||
for data in t("music_level").members() {
|
||||
info[format!("{}_{}", data["masterMusicId"], data["level"])] = data.clone();
|
||||
}
|
||||
info
|
||||
};
|
||||
|
||||
// multievent_rankbonus keyed "{playerCount}_{liveRank}" — _eventPtBonus is a
|
||||
// ratio in 1/10000 (the client renders these as `sum / 100` percent).
|
||||
pub static ref MULTIEVENT_RANK_BONUS: JsonValue = {
|
||||
let mut info = object! {};
|
||||
for data in t("multievent_rankbonus").members() {
|
||||
info[format!("{}_{}", data["playerCount"], data["liveRank"])] = data.clone();
|
||||
}
|
||||
info
|
||||
};
|
||||
|
||||
pub static ref RANKS: JsonValue = t("user_rank");
|
||||
|
||||
pub static ref USER_RANK_REWARD: JsonValue = {
|
||||
|
||||
@@ -25,7 +25,39 @@ pub fn routes(cfg: &mut web::ServiceConfig) {
|
||||
);
|
||||
}
|
||||
|
||||
fn get_event_data(key: &str, event_id: u32) -> JsonValue {
|
||||
// Whether a release_label window is open at `now`. A blank _openedAt has always been
|
||||
// open and a blank _closedAt never closes, which is how the evergreen label (id 1) is
|
||||
// expressed. _releaseStatus other than 1 is never released at all.
|
||||
pub fn release_label_is_open(label_id: i64, now: u64) -> bool {
|
||||
let label = &databases::RELEASE_LABEL[label_id.to_string()];
|
||||
if label.is_empty() || label["releaseStatus"].as_i64().unwrap_or(0) != 1 {
|
||||
return false;
|
||||
}
|
||||
if let Some(opened) = global::parse_datetime(&label["openedAt"].to_string()) {
|
||||
if now < opened {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
if let Some(closed) = global::parse_datetime(&label["closedAt"].to_string()) {
|
||||
if now > closed {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
true
|
||||
}
|
||||
|
||||
// Whether an event is currently running, by its own release label window. ew had no
|
||||
// in-session test before this — nothing else evaluates release_label — so this is the
|
||||
// one place to extend if the event listing ever needs the same question answered.
|
||||
pub fn is_in_session(event_id: u32, now: u64) -> bool {
|
||||
let event = &databases::EVENTS[event_id.to_string()];
|
||||
if event.is_empty() {
|
||||
return false;
|
||||
}
|
||||
release_label_is_open(event["masterReleaseLabelId"].as_i64().unwrap_or(0), now)
|
||||
}
|
||||
|
||||
pub fn get_event_data(key: &str, event_id: u32) -> JsonValue {
|
||||
let mut event = userdata::get_acc_event(key);
|
||||
let is_star_event = STAR_EVENT_IDS.contains(&event_id);
|
||||
//println!("is_star_event: {}, {}", is_star_event, event_id);
|
||||
@@ -50,10 +82,61 @@ fn get_event_data(key: &str, event_id: u32) -> JsonValue {
|
||||
event[event_id.to_string()]["star_event"]["star_event_bonus_daily_count"] = 0.into();
|
||||
}
|
||||
|
||||
normalise_event_shape(&mut event[event_id.to_string()]);
|
||||
|
||||
event[event_id.to_string()].clone()
|
||||
}
|
||||
|
||||
fn save_event_data(key: &str, event_id: u32, data: JsonValue) {
|
||||
// The client's /api/event response types are [Serializable] C# classes, and Unity's
|
||||
// JsonUtility maps them structurally: `score_ranking`, `member_ranking` and `lottery_box`
|
||||
// are OBJECTS (Shock.ProtocolData.ScoreRanking / MemberRanking / LotteryBox), not arrays.
|
||||
// new_user_event.json used to seed them as `[]`, which the client cannot bind - and
|
||||
// MngEventData.SendGetEvent's success callback dereferences `r.data.score_ranking` and
|
||||
// friends unguarded, so a mis-shaped payload takes out the whole recv and its `onComplete`
|
||||
// is never called. That callback is the completion of a GATING TaskFlow.Step in
|
||||
// LiveRestartSelectScene.ReloadScene, so the scene hangs on its loading screen forever.
|
||||
//
|
||||
// The template is fixed, but accounts created before that keep their stored blob, so the
|
||||
// shapes are normalised on the way out as well. Coercion only ever replaces a value the
|
||||
// client could not have parsed anyway; a well-formed object is left exactly as it is.
|
||||
// `set_member` already writes member_ranking as an object, which is the shape this agrees
|
||||
// with.
|
||||
fn normalise_event_shape(event: &mut JsonValue) {
|
||||
fn ensure_object(slot: &mut JsonValue, template: JsonValue) {
|
||||
if !slot.is_object() {
|
||||
*slot = template;
|
||||
} else {
|
||||
for (key, value) in template.entries() {
|
||||
if slot[key].is_null() {
|
||||
slot[key] = value.clone();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ensure_object(&mut event["point_ranking"], object! { rank: 0, point: 0 });
|
||||
ensure_object(&mut event["score_ranking"], object! { all_rank: 0, group_rank: 0, score: 0 });
|
||||
ensure_object(
|
||||
&mut event["member_ranking"],
|
||||
object! { master_character_id: 0, rank: 0, point: 0 },
|
||||
);
|
||||
ensure_object(
|
||||
&mut event["lottery_box"],
|
||||
object! { master_lottery_id: 0, reset_count: 0, draw_count_list: array![] },
|
||||
);
|
||||
// Read by EventData.SetMultiParameter (help count, penalty window, disconnect flag);
|
||||
// absent keys bind as 0/false, but sending them keeps the payload self-describing.
|
||||
for key in ["is_disconnected", "help_count", "penalty_remaining_time"] {
|
||||
if event[key].is_null() {
|
||||
event[key] = 0.into();
|
||||
}
|
||||
}
|
||||
if !event["mission_list"].is_array() {
|
||||
event["mission_list"] = array![];
|
||||
}
|
||||
}
|
||||
|
||||
pub fn save_event_data(key: &str, event_id: u32, data: JsonValue) {
|
||||
let mut event = userdata::get_acc_event(key);
|
||||
|
||||
// Check for old version of event data
|
||||
@@ -209,7 +292,7 @@ async fn set_member(Session { key, body }: Session) -> impl Responder {
|
||||
}))
|
||||
}
|
||||
|
||||
fn get_rank(event: u32, user_id: u64) -> u32 {
|
||||
pub fn get_rank(event: u32, user_id: u64) -> u32 {
|
||||
let scores = crate::router::event_ranking::get_raw_info(event);
|
||||
|
||||
let mut i=1;
|
||||
@@ -254,10 +337,14 @@ fn get_star_rank(points: i64) -> i64 {
|
||||
|
||||
const LIMIT_COINS: i64 = 2000000000;
|
||||
|
||||
fn give_event_points(event_id: u32, amount: i64, user: &mut JsonValue) -> bool {
|
||||
// The row is keyed by BOTH the event and the point type, exactly as get_points reads it
|
||||
// back. Matching on the type alone credited whichever event's row happened to come first
|
||||
// in the list — an account that had ever played a different event got its points there,
|
||||
// and get_points (which does check the id) then reported 0 for the event just played.
|
||||
pub fn give_event_points(event_id: u32, amount: i64, user: &mut JsonValue) -> bool {
|
||||
let mut has = false;
|
||||
for data in user["event_point_list"].members_mut() {
|
||||
if data["type"] == 1 {
|
||||
if data["type"] == 1 && data["master_event_id"] == event_id {
|
||||
has = true;
|
||||
let new_amount = data["amount"].as_i64().unwrap() + amount;
|
||||
if new_amount > LIMIT_COINS {
|
||||
@@ -278,7 +365,7 @@ fn give_event_points(event_id: u32, amount: i64, user: &mut JsonValue) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
fn get_points(event_id: u32, user: &JsonValue) -> i64 {
|
||||
pub fn get_points(event_id: u32, user: &JsonValue) -> i64 {
|
||||
for data in user["event_point_list"].members() {
|
||||
if data["type"] == 1 && data["master_event_id"] == event_id {
|
||||
return data["amount"].as_i64().unwrap()
|
||||
@@ -366,3 +453,96 @@ async fn event_end(req: HttpRequest, Session { key, body }: Session) -> impl Res
|
||||
async fn event_skip(req: HttpRequest, Session { key, body }: Session) -> impl Responder {
|
||||
Api(event_live(&req, &key, &body, true))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
// The client binds these with Unity's JsonUtility against [Serializable] classes:
|
||||
// ScoreRanking { all_rank, group_rank, score }, MemberRanking { master_character_id,
|
||||
// rank, point }, PointRanking { rank, point }, LotteryBox { master_lottery_id,
|
||||
// reset_count, draw_count_list }. An array where an object is expected cannot bind, and
|
||||
// MngEventData.SendGetEvent's callback dereferences them unguarded.
|
||||
#[test]
|
||||
fn the_new_user_template_already_has_the_client_shapes() {
|
||||
let template: JsonValue =
|
||||
jzon::parse(&include_file!("src/router/userdata/new_user_event.json")).unwrap();
|
||||
for key in ["point_ranking", "score_ranking", "member_ranking", "lottery_box"] {
|
||||
assert!(template[key].is_object(), "{} must be an object", key);
|
||||
}
|
||||
assert!(template["mission_list"].is_array());
|
||||
assert_eq!(template["score_ranking"]["all_rank"], 0);
|
||||
assert_eq!(template["member_ranking"]["master_character_id"], 0);
|
||||
assert_eq!(template["lottery_box"]["draw_count_list"], array![]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stored_blobs_with_the_old_array_shapes_are_coerced() {
|
||||
// Exactly what accounts created before the template fix have on disk.
|
||||
let mut stored = object! {
|
||||
point_ranking: object! { point: 12 },
|
||||
score_ranking: array![],
|
||||
member_ranking: array![],
|
||||
lottery_box: array![],
|
||||
mission_list: array![],
|
||||
};
|
||||
normalise_event_shape(&mut stored);
|
||||
|
||||
assert!(stored["score_ranking"].is_object());
|
||||
assert_eq!(stored["score_ranking"]["all_rank"], 0);
|
||||
assert!(stored["member_ranking"].is_object());
|
||||
assert!(stored["lottery_box"].is_object());
|
||||
assert_eq!(stored["lottery_box"]["draw_count_list"], array![]);
|
||||
// A key that was already there survives; the missing sibling is filled in.
|
||||
assert_eq!(stored["point_ranking"]["point"], 12);
|
||||
assert_eq!(stored["point_ranking"]["rank"], 0);
|
||||
// The multi trio EventData.SetMultiParameter reads.
|
||||
assert_eq!(stored["is_disconnected"], 0);
|
||||
assert_eq!(stored["help_count"], 0);
|
||||
assert_eq!(stored["penalty_remaining_time"], 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn well_formed_data_is_left_alone() {
|
||||
let mut live = object! {
|
||||
point_ranking: object! { rank: 3, point: 900 },
|
||||
score_ranking: object! { all_rank: 7, group_rank: 2, score: 4242 },
|
||||
member_ranking: object! { master_character_id: 5, rank: 1, point: 10 },
|
||||
lottery_box: object! { master_lottery_id: 8, reset_count: 1, draw_count_list: array![] },
|
||||
mission_list: array![],
|
||||
is_disconnected: 1,
|
||||
help_count: 4,
|
||||
penalty_remaining_time: 600,
|
||||
};
|
||||
let before = live.clone();
|
||||
normalise_event_shape(&mut live);
|
||||
assert_eq!(live, before);
|
||||
}
|
||||
|
||||
// give_event_points and get_points must agree on what identifies a row. They did not:
|
||||
// the write matched on the point type alone, so an account that had ever earned points
|
||||
// in ANY event credited every later event to that first row — and get_points, which
|
||||
// does compare the event id, then reported 0 for the event actually played.
|
||||
#[test]
|
||||
fn event_points_land_in_the_row_for_that_event() {
|
||||
let mut user = object!{ event_point_list: array![] };
|
||||
|
||||
give_event_points(108, 35, &mut user);
|
||||
assert_eq!(get_points(108, &user), 35);
|
||||
|
||||
// A second event opens a row of its own and leaves the first one alone.
|
||||
give_event_points(111, 70, &mut user);
|
||||
assert_eq!(get_points(111, &user), 70);
|
||||
assert_eq!(get_points(108, &user), 35);
|
||||
assert_eq!(user["event_point_list"].len(), 2);
|
||||
|
||||
// And a second live in the first event still adds to the first row.
|
||||
give_event_points(108, 35, &mut user);
|
||||
assert_eq!(get_points(108, &user), 70);
|
||||
assert_eq!(get_points(111, &user), 70);
|
||||
assert_eq!(user["event_point_list"].len(), 2);
|
||||
|
||||
// An event never played is worth nothing, not somebody else's total.
|
||||
assert_eq!(get_points(115, &user), 0);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -23,10 +23,10 @@ struct AssetVersion {
|
||||
|
||||
static ASSET_VERSIONS: &[AssetVersion] = &[
|
||||
// Default / stock
|
||||
AssetVersion { region: "JP", platform: "Android", version: "4c921d2443335e574a82e04ec9ea243c", hash: "67f8f261c16b3cca63e520a25aad6c1c", latest: true },
|
||||
AssetVersion { region: "JP", platform: "iOS", version: "4c921d2443335e574a82e04ec9ea243c", hash: "b8975be8300013a168d061d3fdcd4a16", latest: true },
|
||||
AssetVersion { region: "GL", platform: "Android", version: "5260ff15dff8ba0c00ad91400f515f55", hash: "d210b28037885f3ef56b8f8aa45ac95b", latest: true },
|
||||
AssetVersion { region: "GL", platform: "iOS", version: "5260ff15dff8ba0c00ad91400f515f55", hash: "dd7175e4bcdab476f38c33c7f34b5e4d", latest: true },
|
||||
AssetVersion { region: "JP", platform: "Android", version: "4c921d2443335e574a82e04ec9ea243c", hash: "67f8f261c16b3cca63e520a25aad6c1c", latest: false },
|
||||
AssetVersion { region: "JP", platform: "iOS", version: "4c921d2443335e574a82e04ec9ea243c", hash: "b8975be8300013a168d061d3fdcd4a16", latest: false },
|
||||
AssetVersion { region: "GL", platform: "Android", version: "5260ff15dff8ba0c00ad91400f515f55", hash: "d210b28037885f3ef56b8f8aa45ac95b", latest: false },
|
||||
AssetVersion { region: "GL", platform: "iOS", version: "5260ff15dff8ba0c00ad91400f515f55", hash: "dd7175e4bcdab476f38c33c7f34b5e4d", latest: false },
|
||||
|
||||
// Re-written client versions 2.0.0 - 2.1.2 (windows only)
|
||||
AssetVersion { region: "JP", platform: "Windows", version: "4c921d2443335e574a82e04ec9ea243c", hash: "4ed1d077df2d1b29e17d25d64fb37242", latest: false },
|
||||
@@ -35,11 +35,16 @@ static ASSET_VERSIONS: &[AssetVersion] = &[
|
||||
AssetVersion { region: "JP", platform: "Windows", version: "ced44f266b4e4c8eb05fe417fd5f3d1b", hash: "ec508163f04e0f9e0435b4302011d123", latest: false },
|
||||
AssetVersion { region: "JP", platform: "Android", version: "ced44f266b4e4c8eb05fe417fd5f3d1b", hash: "27eabc1af04fa6e4a727516d2bbdadff", latest: false },
|
||||
|
||||
// Re-written client versions 2.3.0 -
|
||||
AssetVersion { region: "JP", platform: "Windows", version: "1b2fe99a639b4ca491afe1a841f19d56", hash: "fd37b607ca271a6ffe17b1e2046c45ad", latest: true },
|
||||
// Re-written client versions 2.3.0 - 2.3.2
|
||||
AssetVersion { region: "JP", platform: "Windows", version: "1b2fe99a639b4ca491afe1a841f19d56", hash: "fd37b607ca271a6ffe17b1e2046c45ad", latest: false },
|
||||
AssetVersion { region: "JP", platform: "Android", version: "1b2fe99a639b4ca491afe1a841f19d56", hash: "ed8f4b8df9d3935d689e1236a6816b2b", latest: false },
|
||||
AssetVersion { region: "JP", platform: "iOS", version: "1b2fe99a639b4ca491afe1a841f19d56", hash: "407a8fd81cc5f525ad12c957dbefccb2", latest: false },
|
||||
|
||||
// Re-written client versions 2.4.0 -
|
||||
AssetVersion { region: "JP", platform: "Windows", version: "7d9c2e5efa794a048c11bcaf781ace79", hash: "73c12d3a4986013afa1eee65469124f7", latest: true },
|
||||
AssetVersion { region: "JP", platform: "Android", version: "7d9c2e5efa794a048c11bcaf781ace79", hash: "7d98640952f358554306658c2522d7f6", latest: true },
|
||||
AssetVersion { region: "JP", platform: "iOS", version: "7d9c2e5efa794a048c11bcaf781ace79", hash: "7d7c80f2a8e45789c1f391879710bdbc", latest: true },
|
||||
|
||||
//AssetVersion { region: "JP", platform: "WebGL", version: "4c921d2443335e574a82e04ec9ea243c", hash: "e1ff7c74b20c8d216507972b6f24b9df", latest: true },
|
||||
];
|
||||
|
||||
@@ -289,6 +294,48 @@ pub fn format_datetime(time: u64) -> String {
|
||||
format!("{:04}-{:02}-{:02} {:02}:{:02}:{:02}", y, m, d, secs / 3600, (secs % 3600) / 60, secs % 60)
|
||||
}
|
||||
|
||||
// Inverse of civil_from_days (Howard Hinnant's days_from_civil).
|
||||
fn days_from_civil(y: i64, m: i64, d: i64) -> i64 {
|
||||
let y = if m <= 2 { y - 1 } else { y };
|
||||
let era = if y >= 0 { y } else { y - 399 } / 400;
|
||||
let yoe = y - era * 400;
|
||||
let mp = if m > 2 { m - 3 } else { m + 9 };
|
||||
let doy = (153 * mp + 2) / 5 + d - 1;
|
||||
let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
|
||||
era * 146097 + doe - 719468
|
||||
}
|
||||
|
||||
// Parses the datetime shape masterdata uses ("2023/06/19 5:00:00", also tolerating
|
||||
// '-' separators and a missing time part) into the same naive seconds-since-epoch
|
||||
// scale format_datetime prints. Naive on purpose: every consumer compares a
|
||||
// masterdata timestamp against a server timestamp, so both sides share the offset.
|
||||
pub fn parse_datetime(text: &str) -> Option<u64> {
|
||||
let text = text.trim();
|
||||
if text.is_empty() {
|
||||
return None;
|
||||
}
|
||||
let (date, time) = match text.split_once(' ') {
|
||||
Some((date, time)) => (date, time),
|
||||
None => (text, "0:0:0")
|
||||
};
|
||||
|
||||
let mut date_parts = date.split(['/', '-']);
|
||||
let y = date_parts.next()?.trim().parse::<i64>().ok()?;
|
||||
let m = date_parts.next()?.trim().parse::<i64>().ok()?;
|
||||
let d = date_parts.next()?.trim().parse::<i64>().ok()?;
|
||||
if !(1..=12).contains(&m) || !(1..=31).contains(&d) {
|
||||
return None;
|
||||
}
|
||||
|
||||
let mut time_parts = time.split(':');
|
||||
let hh = time_parts.next().unwrap_or("0").trim().parse::<i64>().ok()?;
|
||||
let mm = time_parts.next().unwrap_or("0").trim().parse::<i64>().ok()?;
|
||||
let ss = time_parts.next().unwrap_or("0").trim().parse::<i64>().ok()?;
|
||||
|
||||
let secs = days_from_civil(y, m, d) * 86400 + hh * 3600 + mm * 60 + ss;
|
||||
if secs < 0 { None } else { Some(secs as u64) }
|
||||
}
|
||||
|
||||
fn init_time(current_time: u64, server_data: &mut JsonValue, token: &str, max_time: u64, max: bool) {
|
||||
let mut edited = false;
|
||||
let default_time = 1709272800;
|
||||
|
||||
@@ -27,7 +27,9 @@ pub fn routes(cfg: &mut web::ServiceConfig) {
|
||||
async fn retire(Session { key, body }: Session) -> impl Responder {
|
||||
live_retire(&key, &body);
|
||||
if body["live_score"]["play_time"].as_i64().unwrap_or(0) > 5 {
|
||||
live_completed(body["master_live_id"].as_i64().unwrap(), body["level"].as_i32().unwrap(), true, 0, 0);
|
||||
// Body-derived, so defaulted rather than unwrapped: a retire is not worth a
|
||||
// panicked worker either (live_completed ignores an unknown id/level).
|
||||
live_completed(body["master_live_id"].as_i64().unwrap_or(0), body["level"].as_i32().unwrap_or(0), true, 0, 0);
|
||||
}
|
||||
Api(Some(object!{
|
||||
"stamina": {},
|
||||
@@ -222,12 +224,15 @@ fn check_for_stale_data(server_data: &mut JsonValue, live_id: i64) {
|
||||
let mut expired = array![];
|
||||
let curr_time = global::timestamp();
|
||||
for (i, live) in server_data["last_live_started"].members().enumerate() {
|
||||
if live["expire_date_time"].as_u64().unwrap() < curr_time || live["master_live_id"] == live_id {
|
||||
// A record with no readable expiry is treated as expired rather than panicking:
|
||||
// start_live always writes one, so this is a corrupt/hand-edited row.
|
||||
let stale = live["expire_date_time"].as_u64().unwrap_or(0) < curr_time;
|
||||
if stale || live["master_live_id"] == live_id {
|
||||
expired.push(i).unwrap();
|
||||
}
|
||||
if live["expire_date_time"].as_u64().unwrap() < curr_time {
|
||||
if stale {
|
||||
// User closed game after losing. Count this as a fail.
|
||||
live_completed(live["master_live_id"].as_i64().unwrap(), live["level"].as_i32().unwrap(), true, 0, 0);
|
||||
live_completed(live["master_live_id"].as_i64().unwrap_or(0), live["level"].as_i32().unwrap_or(0), true, 0, 0);
|
||||
}
|
||||
}
|
||||
for i in expired.members() {
|
||||
@@ -243,34 +248,94 @@ fn get_end_live_deck_id(login_token: &str, body: &JsonValue) -> Option<i32> {
|
||||
let index = server_data["last_live_started"].members().position(|r| r["master_live_id"] == body["master_live_id"])?;
|
||||
let rv = server_data["last_live_started"][index]["deck_slot"].as_i32()?;
|
||||
|
||||
check_for_stale_data(&mut server_data, body["master_live_id"].as_i64().unwrap());
|
||||
check_for_stale_data(&mut server_data, body["master_live_id"].as_i64().unwrap_or(0));
|
||||
userdata::save_server_data(login_token, server_data);
|
||||
Some(rv)
|
||||
}
|
||||
|
||||
pub fn get_end_live_event_id(login_token: &str, body: &JsonValue) -> Option<u32> {
|
||||
// The whole payload the client POSTed to */start for this live, as recorded by
|
||||
// start_live. /multi_live/end needs the boost and deck slot out of it, and its
|
||||
// own request body carries neither.
|
||||
pub fn get_started_live(login_token: &str, body: &JsonValue) -> Option<JsonValue> {
|
||||
let server_data = userdata::get_server_data(login_token);
|
||||
if server_data["last_live_started"].is_null() {
|
||||
return None;
|
||||
}
|
||||
let index = server_data["last_live_started"].members().position(|r| r["master_live_id"] == body["master_live_id"])?;
|
||||
let rv = server_data["last_live_started"][index]["master_event_id"].as_u32()?;
|
||||
|
||||
Some(rv)
|
||||
Some(server_data["last_live_started"][index].clone())
|
||||
}
|
||||
|
||||
fn live_retire(login_token: &str, body: &JsonValue) {
|
||||
// Claims the record start_live left behind: it is returned AND removed in one atomic
|
||||
// step, so of N ends arriving together for one live exactly one gets Some and every other
|
||||
// gets None. /multi_live/end awards off that Some and answers a None from current state,
|
||||
// which is what makes its duplicate protection deliberate rather than a side effect of
|
||||
// get_end_live_deck_id's shape (that one only consumed when the record carried a numeric
|
||||
// deck_slot, and only long after the awarding decision had been taken).
|
||||
//
|
||||
// The stale sweep check_for_stale_data does is folded in unchanged: records past their
|
||||
// hour count as a fail and go, so live_end_ex's own later sweep finds nothing left to
|
||||
// count twice. live_completed is called after the transaction commits - it writes to a
|
||||
// different database and has no business running inside this one's write lock.
|
||||
//
|
||||
// Solo /live/end is deliberately NOT routed through here: it still consumes its record
|
||||
// inside live_end_ex exactly as it always did.
|
||||
pub fn take_started_live(login_token: &str, body: &JsonValue) -> Option<JsonValue> {
|
||||
let live_id = body["master_live_id"].clone();
|
||||
let curr_time = global::timestamp();
|
||||
let mut failed: Vec<(i64, i32)> = Vec::new();
|
||||
|
||||
let taken = userdata::modify_server_data(login_token, |server_data| {
|
||||
if server_data["last_live_started"].is_null() {
|
||||
server_data["last_live_started"] = array![];
|
||||
}
|
||||
let mut taken: Option<JsonValue> = None;
|
||||
let mut kept = array![];
|
||||
for live in server_data["last_live_started"].members() {
|
||||
let stale = live["expire_date_time"].as_u64().unwrap_or(0) < curr_time;
|
||||
let mine = live["master_live_id"] == live_id;
|
||||
if stale {
|
||||
// User closed game after losing. Count this as a fail.
|
||||
failed.push((
|
||||
live["master_live_id"].as_i64().unwrap_or(0),
|
||||
live["level"].as_i32().unwrap_or(0)
|
||||
));
|
||||
}
|
||||
if mine && taken.is_none() {
|
||||
taken = Some(live.clone());
|
||||
} else if !stale && !mine {
|
||||
kept.push(live.clone()).unwrap();
|
||||
}
|
||||
}
|
||||
server_data["last_live_started"] = kept;
|
||||
taken
|
||||
});
|
||||
|
||||
for (id, level) in failed {
|
||||
live_completed(id, level, true, 0, 0);
|
||||
}
|
||||
taken
|
||||
}
|
||||
|
||||
pub fn get_end_live_event_id(login_token: &str, body: &JsonValue) -> Option<u32> {
|
||||
get_started_live(login_token, body)?["master_event_id"].as_u32()
|
||||
}
|
||||
|
||||
pub fn live_retire(login_token: &str, body: &JsonValue) {
|
||||
let mut server_data = userdata::get_server_data(login_token);
|
||||
check_for_stale_data(&mut server_data, body["master_live_id"].as_i64().unwrap());
|
||||
// A body with no master_live_id matches nothing (0 is not a live id); the stale sweep
|
||||
// still runs, which is all a retire with a malformed body can honestly do.
|
||||
check_for_stale_data(&mut server_data, body["master_live_id"].as_i64().unwrap_or(0));
|
||||
userdata::save_server_data(login_token, server_data);
|
||||
}
|
||||
|
||||
fn start_live(login_token: &str, body: &JsonValue) {
|
||||
pub fn start_live(login_token: &str, body: &JsonValue) {
|
||||
let mut server_data = userdata::get_server_data(login_token);
|
||||
if server_data["last_live_started"].is_null() {
|
||||
server_data["last_live_started"] = array![];
|
||||
}
|
||||
check_for_stale_data(&mut server_data, body["master_live_id"].as_i64().unwrap());
|
||||
// Body-derived: /multi_live/start forwards a body this server never validated, and a
|
||||
// start with no live id must record a useless record rather than panic a worker.
|
||||
check_for_stale_data(&mut server_data, body["master_live_id"].as_i64().unwrap_or(0));
|
||||
let mut to_save = body.clone();
|
||||
// The user has 1 hour to complete a live
|
||||
to_save["expire_date_time"] = (global::timestamp() + (1 * 60 * 60)).into();
|
||||
@@ -311,24 +376,32 @@ fn get_clear_count(id: i64, user: &JsonValue) -> i64 {
|
||||
rv
|
||||
}
|
||||
|
||||
pub fn update_live_data(user: &mut JsonValue, data: &JsonValue, add: bool) -> JsonValue {
|
||||
// `record_high_score=false` plays this live for the clear count and the max combo but
|
||||
// leaves the stored high score alone: it enters the keep-best comparison below as 0, so
|
||||
// the existing record always wins and is what gets reported back. That is the official
|
||||
// treatment of a public multi live ("本イベントでは HIGH SCOREは更新されません"), which
|
||||
// /multi_live/end asks for — see the score-board note in live_end_ex.
|
||||
pub fn update_live_data(user: &mut JsonValue, data: &JsonValue, add: bool, record_high_score: bool) -> JsonValue {
|
||||
if user["tutorial_step"].as_i32().unwrap() < 130 {
|
||||
return JsonValue::Null;
|
||||
}
|
||||
|
||||
// Every field here comes off the request body, so a malformed end must not panic a
|
||||
// worker: a missing id/level/score reads as 0, which is what an empty live record
|
||||
// would have held anyway.
|
||||
let mut rv = object!{
|
||||
"master_live_id": data["master_live_id"].as_i64().unwrap(),
|
||||
"level": data["level"].as_i64().unwrap(),
|
||||
"master_live_id": data["master_live_id"].as_i64().unwrap_or(0),
|
||||
"level": data["level"].as_i64().unwrap_or(0),
|
||||
"clear_count": 1,
|
||||
"high_score": data["live_score"]["score"].as_i64().unwrap(),
|
||||
"max_combo": data["live_score"]["max_combo"].as_i64().unwrap(),
|
||||
"high_score": if record_high_score { data["live_score"]["score"].as_i64().unwrap_or(0) } else { 0 },
|
||||
"max_combo": data["live_score"]["max_combo"].as_i64().unwrap_or(0),
|
||||
"auto_enable": 1, //whats this?
|
||||
"updated_time": global::timestamp()
|
||||
};
|
||||
|
||||
let mut has = false;
|
||||
for current in user["live_list"].members_mut() {
|
||||
if current["master_live_id"] == rv["master_live_id"] && (current["level"] == rv["level"] || data["level"].as_i32().unwrap() == 0) {
|
||||
if current["master_live_id"] == rv["master_live_id"] && (current["level"] == rv["level"] || data["level"].as_i32().unwrap_or(0) == 0) {
|
||||
has = true;
|
||||
if add {
|
||||
rv["clear_count"] = (current["clear_count"].as_i64().unwrap() + 1).into();
|
||||
@@ -348,7 +421,7 @@ pub fn update_live_data(user: &mut JsonValue, data: &JsonValue, add: bool) -> Js
|
||||
}
|
||||
current["updated_time"] = rv["updated_time"].clone();
|
||||
rv["level"] = current["level"].clone();
|
||||
if data["level"].as_i32().unwrap() != 0 {
|
||||
if data["level"].as_i32().unwrap_or(0) != 0 {
|
||||
break;
|
||||
}
|
||||
}
|
||||
@@ -586,22 +659,46 @@ fn get_live_character_list(lp_used: i32, deck_id: i32, user: &mut JsonValue, mis
|
||||
}
|
||||
|
||||
pub fn live_end(req: &HttpRequest, key: &str, body: &JsonValue, skipped: bool) -> JsonValue {
|
||||
live_end_ex(req, key, body, skipped, true, true)
|
||||
}
|
||||
|
||||
// consume_lp=false is for lives whose stamina was already taken up front
|
||||
// (/multi_live/start does that, because its response reports consumed_stamina).
|
||||
// Everything else still scales off lp_used, so the caller injects use_lp.
|
||||
//
|
||||
// update_score_board=false plays the live without recording a score anywhere: neither the
|
||||
// account's own high score for the song nor the per-song board /live/ranking serves. Only
|
||||
// /multi_live/end passes false, and only for a PUBLIC (random matchmaking) room — see the
|
||||
// privacy note there. Everything else about the live is untouched: the clear count, the
|
||||
// max combo, the clear-rate counters, missions and every reward are computed off this
|
||||
// play's own score exactly as they always were.
|
||||
pub fn live_end_ex(req: &HttpRequest, key: &str, body: &JsonValue, skipped: bool, consume_lp: bool, update_score_board: bool) -> JsonValue {
|
||||
let mut user2 = userdata::get_acc_home(&key);
|
||||
let mut user = userdata::get_acc(&key);
|
||||
let mut user_missions = userdata::get_acc_missions(&key);
|
||||
let mut chats = userdata::get_acc_chats(&key);
|
||||
|
||||
// Read once, off the request, with a default each: a client (or a replayed/forged
|
||||
// POST) that omits a field must get a wrong-but-harmless result, not a panicked
|
||||
// worker. /multi_live/end is the reachable path — it forwards a body this handler
|
||||
// never validated — but the solo path gets the same treatment, and for a well-formed
|
||||
// body every one of these is exactly what the old unwrap produced.
|
||||
let live_id = body["master_live_id"].as_i64().unwrap_or(0);
|
||||
let level = body["level"].as_i64().unwrap_or(0);
|
||||
let score = body["live_score"]["score"].as_i64().unwrap_or(0);
|
||||
let max_combo = body["live_score"]["max_combo"].as_i64().unwrap_or(0);
|
||||
|
||||
let jp = items::get_region(req.headers());
|
||||
let first_clear = !skipped
|
||||
&& user["tutorial_step"].as_i32().unwrap() >= 130
|
||||
&& get_clear_count(body["master_live_id"].as_i64().unwrap(), &user) == 0;
|
||||
&& get_clear_count(live_id, &user) == 0;
|
||||
|
||||
let live = if skipped {
|
||||
items::use_item(&object!{
|
||||
value: 21000001,
|
||||
amount: 1,
|
||||
consumeType: 4
|
||||
}, body["live_boost"].as_i64().unwrap(), &mut user);
|
||||
}, body["live_boost"].as_i64().unwrap_or(0), &mut user);
|
||||
update_live_data(&mut user, &object!{
|
||||
master_live_id: body["master_live_id"].clone(),
|
||||
level: 0,
|
||||
@@ -609,9 +706,9 @@ pub fn live_end(req: &HttpRequest, key: &str, body: &JsonValue, skipped: bool) -
|
||||
score: 1,
|
||||
max_combo: 1
|
||||
}
|
||||
}, false)
|
||||
}, false, update_score_board)
|
||||
} else {
|
||||
update_live_data(&mut user, &body, true)
|
||||
update_live_data(&mut user, &body, true, update_score_board)
|
||||
};
|
||||
|
||||
//1273009, 1273010, 1273011, 1273012
|
||||
@@ -626,21 +723,27 @@ pub fn live_end(req: &HttpRequest, key: &str, body: &JsonValue, skipped: bool) -
|
||||
}
|
||||
}
|
||||
|
||||
// The account the per-song board is keyed by, or 0 to leave the board alone —
|
||||
// clear_rate::update_live_score's own "no user, no board entry" guard, which is how
|
||||
// /live/retire already counts a play it must not rank. The clear-rate counters still
|
||||
// get the play either way: a public multi live really was played.
|
||||
let board_uid = if update_score_board { user["user"]["id"].as_i64().unwrap() } else { 0 };
|
||||
|
||||
let missions;
|
||||
if skipped {
|
||||
live_completed(body["master_live_id"].as_i64().unwrap(), live["level"].as_i32().unwrap(), false, live["high_score"].as_i64().unwrap(), user["user"]["id"].as_i64().unwrap());
|
||||
let clear_count = get_clear_count(body["master_live_id"].as_i64().unwrap(), &user);
|
||||
live_completed(live_id, live["level"].as_i32().unwrap_or(0), false, live["high_score"].as_i64().unwrap_or(0), board_uid);
|
||||
let clear_count = get_clear_count(live_id, &user);
|
||||
|
||||
missions = get_live_mission_completed_ids(&user, body["master_live_id"].as_i64().unwrap(), live["high_score"].as_i64().unwrap(), live["max_combo"].as_i64().unwrap(), clear_count, live["level"].as_i64().unwrap(), false, false).unwrap_or(array![]);
|
||||
missions = get_live_mission_completed_ids(&user, live_id, live["high_score"].as_i64().unwrap_or(0), live["max_combo"].as_i64().unwrap_or(0), clear_count, live["level"].as_i64().unwrap_or(0), false, false).unwrap_or(array![]);
|
||||
} else {
|
||||
live_completed(body["master_live_id"].as_i64().unwrap(), body["level"].as_i32().unwrap(), false, body["live_score"]["score"].as_i64().unwrap(), user["user"]["id"].as_i64().unwrap());
|
||||
let clear_count = get_clear_count(body["master_live_id"].as_i64().unwrap(), &user);
|
||||
live_completed(live_id, level as i32, false, score, board_uid);
|
||||
let clear_count = get_clear_count(live_id, &user);
|
||||
|
||||
let is_full_combo = (body["live_score"]["good"].as_i32().unwrap_or(1) + body["live_score"]["bad"].as_i32().unwrap_or(1) + body["live_score"]["miss"].as_i32().unwrap_or(1)) == 0;
|
||||
|
||||
let is_perfect = is_full_combo && body["live_score"]["great"].as_i32().unwrap_or(1) == 0;
|
||||
|
||||
missions = get_live_mission_completed_ids(&user, body["master_live_id"].as_i64().unwrap(), body["live_score"]["score"].as_i64().unwrap(), body["live_score"]["max_combo"].as_i64().unwrap(), clear_count, body["level"].as_i64().unwrap(), is_full_combo, is_perfect).unwrap_or(array![]);
|
||||
missions = get_live_mission_completed_ids(&user, live_id, score, max_combo, clear_count, level, is_full_combo, is_perfect).unwrap_or(array![]);
|
||||
|
||||
if is_full_combo {
|
||||
if items::advance_mission(1176001, 1, 1, &mut user_missions).is_some() {
|
||||
@@ -665,13 +768,13 @@ pub fn live_end(req: &HttpRequest, key: &str, body: &JsonValue, skipped: bool) -
|
||||
if is_perfect && items::advance_mission(1177001, 1, 1, &mut user_missions).is_some() {
|
||||
cleared_missions.push(1177001).unwrap();
|
||||
}
|
||||
if is_perfect && body["level"].as_i32().unwrap() == 4 && items::advance_mission(1177002, 1, 1, &mut user_missions).is_some() {
|
||||
if is_perfect && level == 4 && items::advance_mission(1177002, 1, 1, &mut user_missions).is_some() {
|
||||
cleared_missions.push(1177002).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
update_live_mission_data(&mut user, &object!{
|
||||
master_live_id: body["master_live_id"].as_i64().unwrap(),
|
||||
master_live_id: live_id,
|
||||
clear_master_live_mission_ids: missions.clone()
|
||||
});
|
||||
|
||||
@@ -679,7 +782,9 @@ pub fn live_end(req: &HttpRequest, key: &str, body: &JsonValue, skipped: bool) -
|
||||
|
||||
let mut reward_list = give_mission_rewards(&mut user, &mut user2, &missions, &mut user_missions, &mut cleared_missions, &mut chats, (lp_used / 10) as i64, jp);
|
||||
|
||||
if consume_lp {
|
||||
items::lp_modification(&mut user, lp_used as u64, true);
|
||||
}
|
||||
|
||||
items::give_exp(lp_used, &mut user, &mut user_missions, &mut cleared_missions);
|
||||
|
||||
@@ -761,7 +866,7 @@ mod tests {
|
||||
master_live_id: live_id,
|
||||
level: level,
|
||||
live_score: { score: score, max_combo: combo }
|
||||
}, true);
|
||||
}, true, true);
|
||||
userdata::save_acc(token, user);
|
||||
}
|
||||
|
||||
@@ -769,6 +874,118 @@ mod tests {
|
||||
get_clear_count(live_id, &userdata::get_acc(token))
|
||||
}
|
||||
|
||||
// The same clear with the score board switched off, which is what live_end_ex does for
|
||||
// a multi live played in a PUBLIC room. Returns the `live` record the client is
|
||||
// answered with.
|
||||
fn record_unscored_clear(token: &str, live_id: i64, level: i64, score: i64, combo: i64) -> JsonValue {
|
||||
let mut user = userdata::get_acc(token);
|
||||
let live = update_live_data(&mut user, &object!{
|
||||
master_live_id: live_id,
|
||||
level: level,
|
||||
live_score: { score: score, max_combo: combo }
|
||||
}, true, false);
|
||||
userdata::save_acc(token, user);
|
||||
live
|
||||
}
|
||||
|
||||
fn stored_live(token: &str, live_id: i64) -> JsonValue {
|
||||
userdata::get_acc(token)["live_list"]
|
||||
.members()
|
||||
.find(|l| l["master_live_id"] == live_id)
|
||||
.cloned()
|
||||
.unwrap_or(JsonValue::Null)
|
||||
}
|
||||
|
||||
// /multi_live/end's score-board branch, at the write it actually gates. live_end_ex
|
||||
// itself cannot be driven from a test (items::get_region reaches the clap parser, which
|
||||
// rejects the harness's own arguments), so the two halves of the gate are pinned where
|
||||
// they live: the account's own high score here, and the per-song board next door in
|
||||
// clear_rate.
|
||||
//
|
||||
// A PRIVATE room is the unchanged path: it records exactly like a solo live, keep-best.
|
||||
#[test]
|
||||
fn a_private_multi_live_records_its_score_keep_best() {
|
||||
let _lock = crate::runtime::lock_test_data_path();
|
||||
let token = "sb_private_room";
|
||||
register_account(token);
|
||||
|
||||
record_clear(token, STOCK_LIVE_ID, 4, 500000, 320);
|
||||
assert_eq!(stored_live(token, STOCK_LIVE_ID)["high_score"], 500000);
|
||||
|
||||
// A better score wins...
|
||||
record_clear(token, STOCK_LIVE_ID, 4, 600000, 340);
|
||||
assert_eq!(stored_live(token, STOCK_LIVE_ID)["high_score"], 600000);
|
||||
|
||||
// ...and a worse one never overwrites it.
|
||||
record_clear(token, STOCK_LIVE_ID, 4, 100000, 20);
|
||||
assert_eq!(stored_live(token, STOCK_LIVE_ID)["high_score"], 600000);
|
||||
assert_eq!(pulled_clear_count(token, STOCK_LIVE_ID), 3);
|
||||
}
|
||||
|
||||
// A PUBLIC room is the official behaviour: the play counts, the score does not.
|
||||
#[test]
|
||||
fn a_public_multi_live_leaves_the_high_score_alone() {
|
||||
let _lock = crate::runtime::lock_test_data_path();
|
||||
let token = "sb_public_room";
|
||||
register_account(token);
|
||||
|
||||
record_clear(token, STOCK_LIVE_ID, 4, 500000, 320);
|
||||
|
||||
// A score that would beat the stored one is not recorded, and the answer the client
|
||||
// gets back reports the record that still stands rather than this play's score.
|
||||
let live = record_unscored_clear(token, STOCK_LIVE_ID, 4, 900000, 400);
|
||||
assert_eq!(live["high_score"], 500000);
|
||||
assert_eq!(stored_live(token, STOCK_LIVE_ID)["high_score"], 500000);
|
||||
|
||||
// Everything else about the live still lands: the clear counts and the combo is a
|
||||
// real one (the client only ever said HIGH SCORE was not updated).
|
||||
assert_eq!(pulled_clear_count(token, STOCK_LIVE_ID), 2);
|
||||
assert_eq!(stored_live(token, STOCK_LIVE_ID)["max_combo"], 400);
|
||||
}
|
||||
|
||||
// The solo path is untouched by the multi branch: live_end — the only entry /live/end
|
||||
// and /live/skip have — calls live_end_ex with update_score_board hardcoded true, so a
|
||||
// solo live writes exactly what it always did.
|
||||
#[test]
|
||||
fn the_solo_live_path_still_records_its_score() {
|
||||
let _lock = crate::runtime::lock_test_data_path();
|
||||
let token = "sb_solo_path";
|
||||
register_account(token);
|
||||
|
||||
record_clear(token, STOCK_LIVE_ID, 4, 450000, 300);
|
||||
assert_eq!(stored_live(token, STOCK_LIVE_ID)["high_score"], 450000);
|
||||
|
||||
// /live/skip's shape: level 0 matches whatever level is stored, its stand-in score
|
||||
// of 1 can never beat the record, and the record is what it answers with.
|
||||
let mut user = userdata::get_acc(token);
|
||||
let live = update_live_data(&mut user, &object!{
|
||||
master_live_id: STOCK_LIVE_ID,
|
||||
level: 0,
|
||||
live_score: { score: 1, max_combo: 1 }
|
||||
}, false, true);
|
||||
userdata::save_acc(token, user);
|
||||
assert_eq!(live["high_score"], 450000);
|
||||
assert_eq!(stored_live(token, STOCK_LIVE_ID)["high_score"], 450000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_public_multi_live_on_a_new_song_records_no_score_at_all() {
|
||||
let _lock = crate::runtime::lock_test_data_path();
|
||||
let token = "sb_public_first_play";
|
||||
register_account(token);
|
||||
|
||||
// The first ever play of the song is a public multi: the song gets a live record
|
||||
// so the clear counts, but there is no high score to show for it.
|
||||
let live = record_unscored_clear(token, STOCK_LIVE_ID, 4, 900000, 400);
|
||||
assert_eq!(live["high_score"], 0);
|
||||
assert_eq!(stored_live(token, STOCK_LIVE_ID)["high_score"], 0);
|
||||
assert_eq!(pulled_clear_count(token, STOCK_LIVE_ID), 1);
|
||||
|
||||
// A later solo play sets it for real.
|
||||
record_clear(token, STOCK_LIVE_ID, 4, 300000, 100);
|
||||
assert_eq!(stored_live(token, STOCK_LIVE_ID)["high_score"], 300000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn clear_count_persists_with_custom_songs_disabled() {
|
||||
let _lock = crate::runtime::lock_test_data_path();
|
||||
|
||||
1014
src/router/multi_live.rs
Normal file
1014
src/router/multi_live.rs
Normal file
File diff suppressed because it is too large
Load Diff
970
src/router/multi_live/proto.rs
Normal file
970
src/router/multi_live/proto.rs
Normal file
@@ -0,0 +1,970 @@
|
||||
// Wire format for the multi-live WebSocket relay, per docs/multi-live-ws-protocol.md.
|
||||
//
|
||||
// The C# client rewrite encodes/decodes the exact same bytes, so this module is a
|
||||
// straight transcription of the spec tables rather than idiomatic Rust: every field is
|
||||
// written in declaration order, all integers are little-endian, strings are UTF-8 behind
|
||||
// a u16 length, and counts are u8. Nothing here is serde-driven - the type tags mirror
|
||||
// Photon's Hashtable boxing (an Int32 must arrive as Int32 or GetCurrentRoomEventId's
|
||||
// exact `is int` check fails), and a derive would hide that.
|
||||
//
|
||||
// Both directions are implemented in both halves (encode + decode for client messages
|
||||
// AND for server messages) even though the relay only ever decodes the former and
|
||||
// encodes the latter. The unused halves are what the round-trip tests exercise, and they
|
||||
// double as the reference the C# side is written against.
|
||||
|
||||
use std::fmt;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Constants
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// Value tags (see "Value encoding" in the spec).
|
||||
pub const TAG_NULL: u8 = 0;
|
||||
pub const TAG_INT: u8 = 1;
|
||||
pub const TAG_STRING: u8 = 2;
|
||||
pub const TAG_BOOL: u8 = 3;
|
||||
pub const TAG_DOUBLE: u8 = 4;
|
||||
|
||||
// Client -> server opcodes.
|
||||
pub const OP_AUTH: u8 = 1;
|
||||
pub const OP_JOIN_LOBBY: u8 = 2;
|
||||
pub const OP_LEAVE_LOBBY: u8 = 3;
|
||||
pub const OP_CREATE_ROOM: u8 = 4;
|
||||
pub const OP_JOIN_ROOM: u8 = 5;
|
||||
pub const OP_JOIN_RANDOM: u8 = 6;
|
||||
pub const OP_LEAVE_ROOM: u8 = 7;
|
||||
pub const OP_REJOIN: u8 = 8;
|
||||
pub const OP_SET_PLAYER_PROPS: u8 = 9;
|
||||
pub const OP_SET_ROOM_PROPS: u8 = 10;
|
||||
pub const OP_RPC: u8 = 11;
|
||||
pub const OP_PING: u8 = 12;
|
||||
|
||||
// Server -> client opcodes.
|
||||
pub const OP_AUTH_OK: u8 = 20;
|
||||
pub const OP_JOINED_LOBBY: u8 = 21;
|
||||
pub const OP_LEFT_LOBBY: u8 = 22;
|
||||
pub const OP_JOINED_ROOM: u8 = 23;
|
||||
pub const OP_CREATE_FAILED: u8 = 24;
|
||||
pub const OP_JOIN_FAILED: u8 = 25;
|
||||
pub const OP_PLAYER_ENTERED: u8 = 26;
|
||||
pub const OP_PLAYER_LEFT: u8 = 27;
|
||||
pub const OP_LEFT_ROOM: u8 = 28;
|
||||
pub const OP_ROOM_PROPS_CHANGED: u8 = 29;
|
||||
pub const OP_PLAYER_PROPS_CHANGED: u8 = 30;
|
||||
pub const OP_MASTER_SWITCHED: u8 = 31;
|
||||
// Distinct name from the client-side OP_RPC (11); same message, opposite direction.
|
||||
pub const OP_RPC_BROADCAST: u8 = 32;
|
||||
pub const OP_PONG: u8 = 33;
|
||||
pub const OP_KICKED: u8 = 34;
|
||||
|
||||
// Close codes. The spec names 4001 and 4002; 4000 covers the framing errors it
|
||||
// describes ("Text frames are a protocol error -> close", malformed frame, unknown
|
||||
// opcode) without naming a code for them.
|
||||
pub const CLOSE_PROTOCOL: u16 = 4000;
|
||||
pub const CLOSE_UNAUTHENTICATED: u16 = 4001;
|
||||
pub const CLOSE_RATE_LIMIT: u16 = 4002;
|
||||
// Added with the liveness sweep: nothing was heard from this connection for three ping
|
||||
// intervals, so the relay stops holding its seat. See LIVENESS_TIMEOUT in ws.rs.
|
||||
pub const CLOSE_IDLE_TIMEOUT: u16 = 4003;
|
||||
|
||||
// Photon ErrorCode mirrors, so the client's existing logging keeps its meaning.
|
||||
pub const ERR_GAME_ID_ALREADY_EXISTS: i32 = 32766;
|
||||
pub const ERR_GAME_FULL: i32 = 32765;
|
||||
pub const ERR_GAME_CLOSED: i32 = 32764;
|
||||
pub const ERR_NO_RANDOM_MATCH_FOUND: i32 = 32760;
|
||||
pub const ERR_GAME_DOES_NOT_EXIST: i32 = 32758;
|
||||
|
||||
// Kicked causes. A room only dies once nobody is left to be told, so cause 1 is still
|
||||
// reserved — encoded, decoded and tested so the C# side can be written against it now.
|
||||
#[allow(dead_code)]
|
||||
pub const KICK_ROOM_DESTROYED: i32 = 1;
|
||||
// Cause 2 is real: the liveness sweep sends it immediately before close 4003, so a client
|
||||
// that is alive enough to read (a suspended app coming back, a stalled network) can say
|
||||
// why it was dropped instead of showing a bare socket error.
|
||||
pub const KICK_IDLE_TIMEOUT: i32 = 2;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Values and maps
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub enum Value {
|
||||
Null,
|
||||
Int(i32),
|
||||
Str(String),
|
||||
Bool(bool),
|
||||
Double(f64),
|
||||
}
|
||||
|
||||
impl Value {
|
||||
pub fn as_int(&self) -> Option<i32> {
|
||||
match self {
|
||||
Value::Int(v) => Some(*v),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn as_str(&self) -> Option<&str> {
|
||||
match self {
|
||||
Value::Str(v) => Some(v),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// An ordered key -> value bag. Ordering is not semantically meaningful on the wire, but
|
||||
// keeping insertion order makes the changed-subset broadcasts arrive in the order the
|
||||
// sender wrote them, which is one less thing for the client to reason about (and makes
|
||||
// the tests deterministic).
|
||||
#[derive(Clone, Debug, Default, PartialEq)]
|
||||
pub struct Map(Vec<(String, Value)>);
|
||||
|
||||
impl Map {
|
||||
pub fn new() -> Self {
|
||||
Map(Vec::new())
|
||||
}
|
||||
|
||||
// Duplicate keys collapse last-writer-wins, exactly like the Hashtable this mirrors.
|
||||
// Convenience for the tests and for any caller building a bag from scratch; the
|
||||
// relay itself only ever merges into an existing bag.
|
||||
#[allow(dead_code)]
|
||||
pub fn from_pairs<I, K>(pairs: I) -> Self
|
||||
where
|
||||
I: IntoIterator<Item = (K, Value)>,
|
||||
K: Into<String>,
|
||||
{
|
||||
let mut map = Map::new();
|
||||
for (key, value) in pairs {
|
||||
map.set(key, value);
|
||||
}
|
||||
map
|
||||
}
|
||||
|
||||
pub fn get(&self, key: &str) -> Option<&Value> {
|
||||
self.0.iter().find(|(k, _)| k == key).map(|(_, v)| v)
|
||||
}
|
||||
|
||||
pub fn get_int(&self, key: &str) -> Option<i32> {
|
||||
self.get(key).and_then(Value::as_int)
|
||||
}
|
||||
|
||||
pub fn get_str(&self, key: &str) -> Option<&str> {
|
||||
self.get(key).and_then(Value::as_str)
|
||||
}
|
||||
|
||||
// Last-writer-wins in place: an existing key keeps its position.
|
||||
pub fn set<K: Into<String>>(&mut self, key: K, value: Value) {
|
||||
let key = key.into();
|
||||
match self.0.iter_mut().find(|(k, _)| *k == key) {
|
||||
Some(slot) => slot.1 = value,
|
||||
None => self.0.push((key, value)),
|
||||
}
|
||||
}
|
||||
|
||||
// Last-writer-wins merge of `other` INTO self. The relay uses it for property updates
|
||||
// and for seeding/merging a joiner's bag out of the seating op's `playerProps`.
|
||||
pub fn merge(&mut self, other: &Map) {
|
||||
for (key, value) in other.iter() {
|
||||
self.set(key, value.clone());
|
||||
}
|
||||
}
|
||||
|
||||
pub fn len(&self) -> usize {
|
||||
self.0.len()
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.0.is_empty()
|
||||
}
|
||||
|
||||
pub fn iter(&self) -> impl Iterator<Item = (&str, &Value)> {
|
||||
self.0.iter().map(|(k, v)| (k.as_str(), v))
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Messages
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub enum ClientMsg {
|
||||
Auth { user_id: String, token: String },
|
||||
JoinLobby { name: String },
|
||||
LeaveLobby,
|
||||
// `props` is the ROOM bag; `player_props` is the joiner's own bag, mirroring Photon's
|
||||
// OpCreateRoom/OpJoinRoom actorProperties. See docs/multi-live-ws-protocol.md,
|
||||
// "Player properties at join time".
|
||||
CreateRoom {
|
||||
name: String,
|
||||
max_players: u8,
|
||||
visible: bool,
|
||||
open: bool,
|
||||
props: Map,
|
||||
lobby_prop_keys: Vec<String>,
|
||||
player_props: Map,
|
||||
},
|
||||
JoinRoom { name: String, player_props: Map },
|
||||
JoinRandom { power_min: i32, power_max: i32, levels: Vec<Value>, player_props: Map },
|
||||
LeaveRoom,
|
||||
Rejoin { name: String, player_props: Map },
|
||||
SetPlayerProps { props: Map },
|
||||
SetRoomProps { props: Map },
|
||||
Rpc { name: String, params: Vec<Value> },
|
||||
Ping,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub enum ServerMsg {
|
||||
AuthOk { actorless_time_ms: f64 },
|
||||
JoinedLobby,
|
||||
LeftLobby,
|
||||
JoinedRoom {
|
||||
room_name: String,
|
||||
your_actor: i32,
|
||||
master_actor: i32,
|
||||
room_props: Map,
|
||||
players: Vec<(i32, Map)>,
|
||||
},
|
||||
CreateFailed { code: i32, msg: String },
|
||||
JoinFailed { code: i32, msg: String },
|
||||
PlayerEntered { actor: i32, props: Map },
|
||||
PlayerLeft { actor: i32, inactive: bool },
|
||||
LeftRoom,
|
||||
RoomPropsChanged { props: Map },
|
||||
PlayerPropsChanged { actor: i32, props: Map },
|
||||
MasterSwitched { new_master_actor: i32 },
|
||||
Rpc { sender_actor: i32, name: String, params: Vec<Value> },
|
||||
Pong { server_time_ms: f64 },
|
||||
// Sent by the liveness sweep with KICK_IDLE_TIMEOUT; KICK_ROOM_DESTROYED is reserved.
|
||||
Kicked { cause: i32 },
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Decoding
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub enum DecodeError {
|
||||
// Frame ran out before the field did.
|
||||
Truncated,
|
||||
// Zero-length frame: there is not even an opcode.
|
||||
Empty,
|
||||
UnknownOp(u8),
|
||||
UnknownTag(u8),
|
||||
BadUtf8,
|
||||
// "One frame = one message" - anything after the message is a framing bug.
|
||||
TrailingBytes,
|
||||
}
|
||||
|
||||
impl fmt::Display for DecodeError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
|
||||
match self {
|
||||
DecodeError::Truncated => write!(f, "truncated frame"),
|
||||
DecodeError::Empty => write!(f, "empty frame"),
|
||||
DecodeError::UnknownOp(op) => write!(f, "unknown opcode {}", op),
|
||||
DecodeError::UnknownTag(tag) => write!(f, "unknown value tag {}", tag),
|
||||
DecodeError::BadUtf8 => write!(f, "string is not valid utf-8"),
|
||||
DecodeError::TrailingBytes => write!(f, "trailing bytes after message"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct Reader<'a> {
|
||||
buf: &'a [u8],
|
||||
pos: usize,
|
||||
}
|
||||
|
||||
impl<'a> Reader<'a> {
|
||||
fn new(buf: &'a [u8]) -> Self {
|
||||
Reader { buf, pos: 0 }
|
||||
}
|
||||
|
||||
fn take(&mut self, n: usize) -> Result<&'a [u8], DecodeError> {
|
||||
let end = self.pos.checked_add(n).ok_or(DecodeError::Truncated)?;
|
||||
if end > self.buf.len() {
|
||||
return Err(DecodeError::Truncated);
|
||||
}
|
||||
let out = &self.buf[self.pos..end];
|
||||
self.pos = end;
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
fn u8(&mut self) -> Result<u8, DecodeError> {
|
||||
Ok(self.take(1)?[0])
|
||||
}
|
||||
|
||||
fn bool(&mut self) -> Result<bool, DecodeError> {
|
||||
// Photon booleans are one byte; anything non-zero is true.
|
||||
Ok(self.u8()? != 0)
|
||||
}
|
||||
|
||||
fn u16(&mut self) -> Result<u16, DecodeError> {
|
||||
let b = self.take(2)?;
|
||||
Ok(u16::from_le_bytes([b[0], b[1]]))
|
||||
}
|
||||
|
||||
fn i32(&mut self) -> Result<i32, DecodeError> {
|
||||
let b = self.take(4)?;
|
||||
Ok(i32::from_le_bytes([b[0], b[1], b[2], b[3]]))
|
||||
}
|
||||
|
||||
fn f64(&mut self) -> Result<f64, DecodeError> {
|
||||
let b = self.take(8)?;
|
||||
Ok(f64::from_le_bytes([b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7]]))
|
||||
}
|
||||
|
||||
fn string(&mut self) -> Result<String, DecodeError> {
|
||||
let len = self.u16()? as usize;
|
||||
let bytes = self.take(len)?;
|
||||
String::from_utf8(bytes.to_vec()).map_err(|_| DecodeError::BadUtf8)
|
||||
}
|
||||
|
||||
fn value(&mut self) -> Result<Value, DecodeError> {
|
||||
let tag = self.u8()?;
|
||||
match tag {
|
||||
TAG_NULL => Ok(Value::Null),
|
||||
TAG_INT => Ok(Value::Int(self.i32()?)),
|
||||
TAG_STRING => Ok(Value::Str(self.string()?)),
|
||||
TAG_BOOL => Ok(Value::Bool(self.bool()?)),
|
||||
TAG_DOUBLE => Ok(Value::Double(self.f64()?)),
|
||||
other => Err(DecodeError::UnknownTag(other)),
|
||||
}
|
||||
}
|
||||
|
||||
fn map(&mut self) -> Result<Map, DecodeError> {
|
||||
let count = self.u8()?;
|
||||
let mut map = Map::new();
|
||||
for _ in 0..count {
|
||||
let key = self.string()?;
|
||||
let value = self.value()?;
|
||||
map.set(key, value);
|
||||
}
|
||||
Ok(map)
|
||||
}
|
||||
|
||||
fn value_list(&mut self) -> Result<Vec<Value>, DecodeError> {
|
||||
let count = self.u8()?;
|
||||
let mut out = Vec::with_capacity(count as usize);
|
||||
for _ in 0..count {
|
||||
out.push(self.value()?);
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
fn string_list(&mut self) -> Result<Vec<String>, DecodeError> {
|
||||
let count = self.u8()?;
|
||||
let mut out = Vec::with_capacity(count as usize);
|
||||
for _ in 0..count {
|
||||
out.push(self.string()?);
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
fn finish<T>(&self, value: T) -> Result<T, DecodeError> {
|
||||
if self.pos != self.buf.len() {
|
||||
return Err(DecodeError::TrailingBytes);
|
||||
}
|
||||
Ok(value)
|
||||
}
|
||||
}
|
||||
|
||||
pub fn decode_client(buf: &[u8]) -> Result<ClientMsg, DecodeError> {
|
||||
let mut r = Reader::new(buf);
|
||||
let op = r.u8().map_err(|_| DecodeError::Empty)?;
|
||||
let msg = match op {
|
||||
OP_AUTH => ClientMsg::Auth { user_id: r.string()?, token: r.string()? },
|
||||
OP_JOIN_LOBBY => ClientMsg::JoinLobby { name: r.string()? },
|
||||
OP_LEAVE_LOBBY => ClientMsg::LeaveLobby,
|
||||
OP_CREATE_ROOM => ClientMsg::CreateRoom {
|
||||
name: r.string()?,
|
||||
max_players: r.u8()?,
|
||||
visible: r.bool()?,
|
||||
open: r.bool()?,
|
||||
props: r.map()?,
|
||||
lobby_prop_keys: r.string_list()?,
|
||||
player_props: r.map()?,
|
||||
},
|
||||
OP_JOIN_ROOM => ClientMsg::JoinRoom { name: r.string()?, player_props: r.map()? },
|
||||
OP_JOIN_RANDOM => ClientMsg::JoinRandom {
|
||||
power_min: r.i32()?,
|
||||
power_max: r.i32()?,
|
||||
levels: r.value_list()?,
|
||||
player_props: r.map()?,
|
||||
},
|
||||
OP_LEAVE_ROOM => ClientMsg::LeaveRoom,
|
||||
OP_REJOIN => ClientMsg::Rejoin { name: r.string()?, player_props: r.map()? },
|
||||
OP_SET_PLAYER_PROPS => ClientMsg::SetPlayerProps { props: r.map()? },
|
||||
OP_SET_ROOM_PROPS => ClientMsg::SetRoomProps { props: r.map()? },
|
||||
OP_RPC => ClientMsg::Rpc { name: r.string()?, params: r.value_list()? },
|
||||
OP_PING => ClientMsg::Ping,
|
||||
other => return Err(DecodeError::UnknownOp(other)),
|
||||
};
|
||||
r.finish(msg)
|
||||
}
|
||||
|
||||
// The client's half of the codec. The relay never calls it; it is the executable
|
||||
// reference the C# rewrite is written against, and what the round-trip tests drive.
|
||||
#[allow(dead_code)]
|
||||
pub fn decode_server(buf: &[u8]) -> Result<ServerMsg, DecodeError> {
|
||||
let mut r = Reader::new(buf);
|
||||
let op = r.u8().map_err(|_| DecodeError::Empty)?;
|
||||
let msg = match op {
|
||||
OP_AUTH_OK => ServerMsg::AuthOk { actorless_time_ms: r.f64()? },
|
||||
OP_JOINED_LOBBY => ServerMsg::JoinedLobby,
|
||||
OP_LEFT_LOBBY => ServerMsg::LeftLobby,
|
||||
OP_JOINED_ROOM => {
|
||||
let room_name = r.string()?;
|
||||
let your_actor = r.i32()?;
|
||||
let master_actor = r.i32()?;
|
||||
let room_props = r.map()?;
|
||||
let count = r.u8()?;
|
||||
let mut players = Vec::with_capacity(count as usize);
|
||||
for _ in 0..count {
|
||||
let actor = r.i32()?;
|
||||
players.push((actor, r.map()?));
|
||||
}
|
||||
ServerMsg::JoinedRoom { room_name, your_actor, master_actor, room_props, players }
|
||||
}
|
||||
OP_CREATE_FAILED => ServerMsg::CreateFailed { code: r.i32()?, msg: r.string()? },
|
||||
OP_JOIN_FAILED => ServerMsg::JoinFailed { code: r.i32()?, msg: r.string()? },
|
||||
OP_PLAYER_ENTERED => ServerMsg::PlayerEntered { actor: r.i32()?, props: r.map()? },
|
||||
OP_PLAYER_LEFT => ServerMsg::PlayerLeft { actor: r.i32()?, inactive: r.bool()? },
|
||||
OP_LEFT_ROOM => ServerMsg::LeftRoom,
|
||||
OP_ROOM_PROPS_CHANGED => ServerMsg::RoomPropsChanged { props: r.map()? },
|
||||
OP_PLAYER_PROPS_CHANGED => {
|
||||
ServerMsg::PlayerPropsChanged { actor: r.i32()?, props: r.map()? }
|
||||
}
|
||||
OP_MASTER_SWITCHED => ServerMsg::MasterSwitched { new_master_actor: r.i32()? },
|
||||
OP_RPC_BROADCAST => ServerMsg::Rpc {
|
||||
sender_actor: r.i32()?,
|
||||
name: r.string()?,
|
||||
params: r.value_list()?,
|
||||
},
|
||||
OP_PONG => ServerMsg::Pong { server_time_ms: r.f64()? },
|
||||
OP_KICKED => ServerMsg::Kicked { cause: r.i32()? },
|
||||
other => return Err(DecodeError::UnknownOp(other)),
|
||||
};
|
||||
r.finish(msg)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Encoding
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// The relay never legitimately produces a string past 64KiB or a count past 255 (keys
|
||||
// are one or two characters, rooms hold four players, RPC params are a handful of ints),
|
||||
// so encoding is infallible and the guards below only exist to keep a bug from emitting
|
||||
// a frame the peer would mis-frame. They log and clamp rather than panic in a handler.
|
||||
fn put_str(out: &mut Vec<u8>, s: &str) {
|
||||
let mut bytes = s.as_bytes();
|
||||
if bytes.len() > u16::MAX as usize {
|
||||
println!("multi_live/ws: string of {} bytes truncated to fit u16 length", bytes.len());
|
||||
let mut end = u16::MAX as usize;
|
||||
while end > 0 && !s.is_char_boundary(end) {
|
||||
end -= 1;
|
||||
}
|
||||
bytes = &s.as_bytes()[..end];
|
||||
}
|
||||
out.extend_from_slice(&(bytes.len() as u16).to_le_bytes());
|
||||
out.extend_from_slice(bytes);
|
||||
}
|
||||
|
||||
fn put_count(out: &mut Vec<u8>, count: usize, what: &str) -> usize {
|
||||
let clamped = if count > u8::MAX as usize {
|
||||
println!("multi_live/ws: {} count {} clamped to 255", what, count);
|
||||
u8::MAX as usize
|
||||
} else {
|
||||
count
|
||||
};
|
||||
out.push(clamped as u8);
|
||||
clamped
|
||||
}
|
||||
|
||||
fn put_value(out: &mut Vec<u8>, value: &Value) {
|
||||
match value {
|
||||
Value::Null => out.push(TAG_NULL),
|
||||
Value::Int(v) => {
|
||||
out.push(TAG_INT);
|
||||
out.extend_from_slice(&v.to_le_bytes());
|
||||
}
|
||||
Value::Str(v) => {
|
||||
out.push(TAG_STRING);
|
||||
put_str(out, v);
|
||||
}
|
||||
Value::Bool(v) => {
|
||||
out.push(TAG_BOOL);
|
||||
out.push(if *v { 1 } else { 0 });
|
||||
}
|
||||
Value::Double(v) => {
|
||||
out.push(TAG_DOUBLE);
|
||||
out.extend_from_slice(&v.to_le_bytes());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn put_map(out: &mut Vec<u8>, map: &Map) {
|
||||
let count = put_count(out, map.len(), "map");
|
||||
for (key, value) in map.iter().take(count) {
|
||||
put_str(out, key);
|
||||
put_value(out, value);
|
||||
}
|
||||
}
|
||||
|
||||
fn put_value_list(out: &mut Vec<u8>, values: &[Value]) {
|
||||
let count = put_count(out, values.len(), "value list");
|
||||
for value in values.iter().take(count) {
|
||||
put_value(out, value);
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
fn put_string_list(out: &mut Vec<u8>, values: &[String]) {
|
||||
let count = put_count(out, values.len(), "string list");
|
||||
for value in values.iter().take(count) {
|
||||
put_str(out, value);
|
||||
}
|
||||
}
|
||||
|
||||
// See decode_server: the client's half, kept honest by the round-trip tests.
|
||||
#[allow(dead_code)]
|
||||
pub fn encode_client(msg: &ClientMsg) -> Vec<u8> {
|
||||
let mut out = Vec::new();
|
||||
match msg {
|
||||
ClientMsg::Auth { user_id, token } => {
|
||||
out.push(OP_AUTH);
|
||||
put_str(&mut out, user_id);
|
||||
put_str(&mut out, token);
|
||||
}
|
||||
ClientMsg::JoinLobby { name } => {
|
||||
out.push(OP_JOIN_LOBBY);
|
||||
put_str(&mut out, name);
|
||||
}
|
||||
ClientMsg::LeaveLobby => out.push(OP_LEAVE_LOBBY),
|
||||
ClientMsg::CreateRoom { name, max_players, visible, open, props, lobby_prop_keys, player_props } => {
|
||||
out.push(OP_CREATE_ROOM);
|
||||
put_str(&mut out, name);
|
||||
out.push(*max_players);
|
||||
out.push(if *visible { 1 } else { 0 });
|
||||
out.push(if *open { 1 } else { 0 });
|
||||
put_map(&mut out, props);
|
||||
put_string_list(&mut out, lobby_prop_keys);
|
||||
put_map(&mut out, player_props);
|
||||
}
|
||||
ClientMsg::JoinRoom { name, player_props } => {
|
||||
out.push(OP_JOIN_ROOM);
|
||||
put_str(&mut out, name);
|
||||
put_map(&mut out, player_props);
|
||||
}
|
||||
ClientMsg::JoinRandom { power_min, power_max, levels, player_props } => {
|
||||
out.push(OP_JOIN_RANDOM);
|
||||
out.extend_from_slice(&power_min.to_le_bytes());
|
||||
out.extend_from_slice(&power_max.to_le_bytes());
|
||||
put_value_list(&mut out, levels);
|
||||
put_map(&mut out, player_props);
|
||||
}
|
||||
ClientMsg::LeaveRoom => out.push(OP_LEAVE_ROOM),
|
||||
ClientMsg::Rejoin { name, player_props } => {
|
||||
out.push(OP_REJOIN);
|
||||
put_str(&mut out, name);
|
||||
put_map(&mut out, player_props);
|
||||
}
|
||||
ClientMsg::SetPlayerProps { props } => {
|
||||
out.push(OP_SET_PLAYER_PROPS);
|
||||
put_map(&mut out, props);
|
||||
}
|
||||
ClientMsg::SetRoomProps { props } => {
|
||||
out.push(OP_SET_ROOM_PROPS);
|
||||
put_map(&mut out, props);
|
||||
}
|
||||
ClientMsg::Rpc { name, params } => {
|
||||
out.push(OP_RPC);
|
||||
put_str(&mut out, name);
|
||||
put_value_list(&mut out, params);
|
||||
}
|
||||
ClientMsg::Ping => out.push(OP_PING),
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
pub fn encode_server(msg: &ServerMsg) -> Vec<u8> {
|
||||
let mut out = Vec::new();
|
||||
match msg {
|
||||
ServerMsg::AuthOk { actorless_time_ms } => {
|
||||
out.push(OP_AUTH_OK);
|
||||
out.extend_from_slice(&actorless_time_ms.to_le_bytes());
|
||||
}
|
||||
ServerMsg::JoinedLobby => out.push(OP_JOINED_LOBBY),
|
||||
ServerMsg::LeftLobby => out.push(OP_LEFT_LOBBY),
|
||||
ServerMsg::JoinedRoom { room_name, your_actor, master_actor, room_props, players } => {
|
||||
out.push(OP_JOINED_ROOM);
|
||||
put_str(&mut out, room_name);
|
||||
out.extend_from_slice(&your_actor.to_le_bytes());
|
||||
out.extend_from_slice(&master_actor.to_le_bytes());
|
||||
put_map(&mut out, room_props);
|
||||
let count = put_count(&mut out, players.len(), "player");
|
||||
for (actor, props) in players.iter().take(count) {
|
||||
out.extend_from_slice(&actor.to_le_bytes());
|
||||
put_map(&mut out, props);
|
||||
}
|
||||
}
|
||||
ServerMsg::CreateFailed { code, msg } => {
|
||||
out.push(OP_CREATE_FAILED);
|
||||
out.extend_from_slice(&code.to_le_bytes());
|
||||
put_str(&mut out, msg);
|
||||
}
|
||||
ServerMsg::JoinFailed { code, msg } => {
|
||||
out.push(OP_JOIN_FAILED);
|
||||
out.extend_from_slice(&code.to_le_bytes());
|
||||
put_str(&mut out, msg);
|
||||
}
|
||||
ServerMsg::PlayerEntered { actor, props } => {
|
||||
out.push(OP_PLAYER_ENTERED);
|
||||
out.extend_from_slice(&actor.to_le_bytes());
|
||||
put_map(&mut out, props);
|
||||
}
|
||||
ServerMsg::PlayerLeft { actor, inactive } => {
|
||||
out.push(OP_PLAYER_LEFT);
|
||||
out.extend_from_slice(&actor.to_le_bytes());
|
||||
out.push(if *inactive { 1 } else { 0 });
|
||||
}
|
||||
ServerMsg::LeftRoom => out.push(OP_LEFT_ROOM),
|
||||
ServerMsg::RoomPropsChanged { props } => {
|
||||
out.push(OP_ROOM_PROPS_CHANGED);
|
||||
put_map(&mut out, props);
|
||||
}
|
||||
ServerMsg::PlayerPropsChanged { actor, props } => {
|
||||
out.push(OP_PLAYER_PROPS_CHANGED);
|
||||
out.extend_from_slice(&actor.to_le_bytes());
|
||||
put_map(&mut out, props);
|
||||
}
|
||||
ServerMsg::MasterSwitched { new_master_actor } => {
|
||||
out.push(OP_MASTER_SWITCHED);
|
||||
out.extend_from_slice(&new_master_actor.to_le_bytes());
|
||||
}
|
||||
ServerMsg::Rpc { sender_actor, name, params } => {
|
||||
out.push(OP_RPC_BROADCAST);
|
||||
out.extend_from_slice(&sender_actor.to_le_bytes());
|
||||
put_str(&mut out, name);
|
||||
put_value_list(&mut out, params);
|
||||
}
|
||||
ServerMsg::Pong { server_time_ms } => {
|
||||
out.push(OP_PONG);
|
||||
out.extend_from_slice(&server_time_ms.to_le_bytes());
|
||||
}
|
||||
ServerMsg::Kicked { cause } => {
|
||||
out.push(OP_KICKED);
|
||||
out.extend_from_slice(&cause.to_le_bytes());
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn round_client(msg: ClientMsg) {
|
||||
let bytes = encode_client(&msg);
|
||||
let back = decode_client(&bytes).expect("client message decodes");
|
||||
assert_eq!(back, msg);
|
||||
assert_eq!(encode_client(&back), bytes);
|
||||
}
|
||||
|
||||
fn round_server(msg: ServerMsg) {
|
||||
let bytes = encode_server(&msg);
|
||||
let back = decode_server(&bytes).expect("server message decodes");
|
||||
assert_eq!(back, msg);
|
||||
assert_eq!(encode_server(&back), bytes);
|
||||
}
|
||||
|
||||
fn sample_map() -> Map {
|
||||
Map::from_pairs(vec![
|
||||
("A", Value::Int(-7)),
|
||||
("B", Value::Str("ラブライブ".to_string())),
|
||||
("C", Value::Bool(true)),
|
||||
("D", Value::Double(0.5)),
|
||||
("E", Value::Null),
|
||||
])
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn every_client_message_round_trips() {
|
||||
round_client(ClientMsg::Auth { user_id: "12345".into(), token: "a-uuid".into() });
|
||||
round_client(ClientMsg::JoinLobby { name: "MultiEventLobby_31".into() });
|
||||
round_client(ClientMsg::LeaveLobby);
|
||||
round_client(ClientMsg::CreateRoom {
|
||||
name: "123456".into(),
|
||||
max_players: 4,
|
||||
visible: false,
|
||||
open: true,
|
||||
props: sample_map(),
|
||||
lobby_prop_keys: vec!["C0".into(), "C1".into()],
|
||||
player_props: sample_map(),
|
||||
});
|
||||
round_client(ClientMsg::JoinRoom {
|
||||
name: "042719".into(),
|
||||
player_props: sample_map(),
|
||||
});
|
||||
round_client(ClientMsg::JoinRandom {
|
||||
power_min: 0,
|
||||
power_max: i32::MAX,
|
||||
levels: vec![Value::Int(1), Value::Int(4)],
|
||||
player_props: sample_map(),
|
||||
});
|
||||
round_client(ClientMsg::LeaveRoom);
|
||||
round_client(ClientMsg::Rejoin {
|
||||
name: "1000000".into(),
|
||||
player_props: sample_map(),
|
||||
});
|
||||
round_client(ClientMsg::SetPlayerProps { props: sample_map() });
|
||||
round_client(ClientMsg::SetRoomProps { props: sample_map() });
|
||||
round_client(ClientMsg::Rpc {
|
||||
name: "SendStamp".into(),
|
||||
params: vec![Value::Int(3), Value::Str("hi".into())],
|
||||
});
|
||||
round_client(ClientMsg::Ping);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn every_server_message_round_trips() {
|
||||
round_server(ServerMsg::AuthOk { actorless_time_ms: 1_754_500_000_123.0 });
|
||||
round_server(ServerMsg::JoinedLobby);
|
||||
round_server(ServerMsg::LeftLobby);
|
||||
round_server(ServerMsg::JoinedRoom {
|
||||
room_name: "1000000".into(),
|
||||
your_actor: 2,
|
||||
master_actor: 1,
|
||||
room_props: sample_map(),
|
||||
players: vec![(1, sample_map()), (2, Map::new())],
|
||||
});
|
||||
round_server(ServerMsg::CreateFailed {
|
||||
code: ERR_GAME_ID_ALREADY_EXISTS,
|
||||
msg: "GameIdAlreadyExists".into(),
|
||||
});
|
||||
round_server(ServerMsg::JoinFailed { code: ERR_GAME_FULL, msg: "GameFull".into() });
|
||||
round_server(ServerMsg::PlayerEntered { actor: 3, props: sample_map() });
|
||||
round_server(ServerMsg::PlayerLeft { actor: 3, inactive: true });
|
||||
round_server(ServerMsg::PlayerLeft { actor: 3, inactive: false });
|
||||
round_server(ServerMsg::LeftRoom);
|
||||
round_server(ServerMsg::RoomPropsChanged { props: sample_map() });
|
||||
round_server(ServerMsg::PlayerPropsChanged { actor: 4, props: sample_map() });
|
||||
round_server(ServerMsg::MasterSwitched { new_master_actor: 2 });
|
||||
round_server(ServerMsg::Rpc {
|
||||
sender_actor: 1,
|
||||
name: "_MoveScene".into(),
|
||||
params: vec![Value::Null],
|
||||
});
|
||||
round_server(ServerMsg::Pong { server_time_ms: -0.0 });
|
||||
round_server(ServerMsg::Kicked { cause: KICK_ROOM_DESTROYED });
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_map_and_empty_strings_round_trip() {
|
||||
round_client(ClientMsg::Auth { user_id: String::new(), token: String::new() });
|
||||
round_client(ClientMsg::JoinLobby { name: String::new() });
|
||||
round_client(ClientMsg::SetPlayerProps { props: Map::new() });
|
||||
round_client(ClientMsg::Rpc { name: String::new(), params: Vec::new() });
|
||||
round_client(ClientMsg::CreateRoom {
|
||||
name: String::new(),
|
||||
max_players: 0,
|
||||
visible: true,
|
||||
open: false,
|
||||
props: Map::new(),
|
||||
lobby_prop_keys: Vec::new(),
|
||||
player_props: Map::new(),
|
||||
});
|
||||
// A joiner that has committed nothing yet sends an empty bag on every seating op.
|
||||
round_client(ClientMsg::JoinRoom { name: String::new(), player_props: Map::new() });
|
||||
round_client(ClientMsg::Rejoin { name: String::new(), player_props: Map::new() });
|
||||
round_client(ClientMsg::JoinRandom {
|
||||
power_min: 0,
|
||||
power_max: 0,
|
||||
levels: Vec::new(),
|
||||
player_props: Map::new(),
|
||||
});
|
||||
// The trailing playerProps of a JoinRoom is exactly one byte when empty.
|
||||
assert_eq!(
|
||||
encode_client(&ClientMsg::JoinRoom { name: String::new(), player_props: Map::new() }),
|
||||
vec![OP_JOIN_ROOM, 0, 0, 0]
|
||||
);
|
||||
round_server(ServerMsg::JoinedRoom {
|
||||
room_name: String::new(),
|
||||
your_actor: 1,
|
||||
master_actor: 1,
|
||||
room_props: Map::new(),
|
||||
players: Vec::new(),
|
||||
});
|
||||
// An empty map is exactly one byte of payload.
|
||||
assert_eq!(encode_client(&ClientMsg::SetRoomProps { props: Map::new() }), vec![OP_SET_ROOM_PROPS, 0]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn max_u8_counts_round_trip() {
|
||||
let map = Map::from_pairs((0..255).map(|i| (format!("k{}", i), Value::Int(i))));
|
||||
assert_eq!(map.len(), 255);
|
||||
round_client(ClientMsg::SetPlayerProps { props: map.clone() });
|
||||
|
||||
let params: Vec<Value> = (0..255).map(Value::Int).collect();
|
||||
round_client(ClientMsg::Rpc { name: "SendStamp".into(), params });
|
||||
|
||||
let keys: Vec<String> = (0..255).map(|i| format!("k{}", i)).collect();
|
||||
round_client(ClientMsg::CreateRoom {
|
||||
name: "n".into(),
|
||||
max_players: 255,
|
||||
visible: true,
|
||||
open: true,
|
||||
props: map.clone(),
|
||||
lobby_prop_keys: keys,
|
||||
player_props: map.clone(),
|
||||
});
|
||||
// A full 255-key bag on each of the other three seating ops.
|
||||
round_client(ClientMsg::JoinRoom { name: "n".into(), player_props: map.clone() });
|
||||
round_client(ClientMsg::Rejoin { name: "n".into(), player_props: map.clone() });
|
||||
round_client(ClientMsg::JoinRandom {
|
||||
power_min: 0,
|
||||
power_max: 1,
|
||||
levels: vec![Value::Int(4)],
|
||||
player_props: map,
|
||||
});
|
||||
|
||||
let players: Vec<(i32, Map)> = (0..255).map(|i| (i, Map::new())).collect();
|
||||
round_server(ServerMsg::JoinedRoom {
|
||||
room_name: "n".into(),
|
||||
your_actor: 1,
|
||||
master_actor: 1,
|
||||
room_props: Map::new(),
|
||||
players,
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn integer_and_double_extremes_survive() {
|
||||
round_client(ClientMsg::JoinRandom {
|
||||
power_min: i32::MIN,
|
||||
power_max: i32::MAX,
|
||||
levels: vec![Value::Int(i32::MIN), Value::Int(i32::MAX), Value::Int(0)],
|
||||
player_props: Map::new(),
|
||||
});
|
||||
round_server(ServerMsg::PlayerEntered {
|
||||
actor: i32::MIN,
|
||||
props: Map::from_pairs(vec![
|
||||
("a", Value::Double(f64::MAX)),
|
||||
("b", Value::Double(f64::MIN_POSITIVE)),
|
||||
("c", Value::Double(f64::INFINITY)),
|
||||
]),
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn wire_layout_is_byte_exact() {
|
||||
// Guard the framing itself: little-endian everywhere, u16 string lengths,
|
||||
// u8 counts, tag-then-payload values. If this test needs updating, the C#
|
||||
// client needs updating too.
|
||||
assert_eq!(
|
||||
encode_client(&ClientMsg::Auth { user_id: "7".into(), token: "ab".into() }),
|
||||
vec![OP_AUTH, 1, 0, b'7', 2, 0, b'a', b'b']
|
||||
);
|
||||
assert_eq!(
|
||||
encode_client(&ClientMsg::JoinRandom {
|
||||
power_min: 1,
|
||||
power_max: 256,
|
||||
levels: vec![Value::Int(4)],
|
||||
player_props: Map::new(),
|
||||
}),
|
||||
// ... levels ValueList ..., then the trailing empty playerProps map (count 0).
|
||||
vec![OP_JOIN_RANDOM, 1, 0, 0, 0, 0, 1, 0, 0, 1, TAG_INT, 4, 0, 0, 0, 0]
|
||||
);
|
||||
// The seating ops' trailing bag is an ordinary Map: count:u8 then (key,value)*.
|
||||
assert_eq!(
|
||||
encode_client(&ClientMsg::JoinRoom {
|
||||
name: "42".into(),
|
||||
player_props: Map::from_pairs(vec![("B", Value::Str("7".into()))]),
|
||||
}),
|
||||
vec![OP_JOIN_ROOM, 2, 0, b'4', b'2', 1, 1, 0, b'B', TAG_STRING, 1, 0, b'7']
|
||||
);
|
||||
assert_eq!(
|
||||
encode_client(&ClientMsg::Rejoin {
|
||||
name: "42".into(),
|
||||
player_props: Map::from_pairs(vec![("F", Value::Int(5))]),
|
||||
}),
|
||||
vec![OP_REJOIN, 2, 0, b'4', b'2', 1, 1, 0, b'F', TAG_INT, 5, 0, 0, 0]
|
||||
);
|
||||
assert_eq!(
|
||||
encode_client(&ClientMsg::SetPlayerProps {
|
||||
props: Map::from_pairs(vec![("LB", Value::Int(12))]),
|
||||
}),
|
||||
vec![OP_SET_PLAYER_PROPS, 1, 2, 0, b'L', b'B', TAG_INT, 12, 0, 0, 0]
|
||||
);
|
||||
assert_eq!(
|
||||
encode_server(&ServerMsg::PlayerLeft { actor: 2, inactive: true }),
|
||||
vec![OP_PLAYER_LEFT, 2, 0, 0, 0, 1]
|
||||
);
|
||||
assert_eq!(
|
||||
encode_server(&ServerMsg::AuthOk { actorless_time_ms: 1.0 }),
|
||||
vec![OP_AUTH_OK, 0, 0, 0, 0, 0, 0, 0xf0, 0x3f]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn malformed_frames_are_rejected() {
|
||||
assert_eq!(decode_client(&[]), Err(DecodeError::Empty));
|
||||
assert_eq!(decode_client(&[99]), Err(DecodeError::UnknownOp(99)));
|
||||
assert_eq!(decode_server(&[99]), Err(DecodeError::UnknownOp(99)));
|
||||
// JoinLobby with a length prefix longer than the payload.
|
||||
assert_eq!(decode_client(&[OP_JOIN_LOBBY, 4, 0, b'a']), Err(DecodeError::Truncated));
|
||||
// Ping carries nothing.
|
||||
assert_eq!(decode_client(&[OP_PING, 0]), Err(DecodeError::TrailingBytes));
|
||||
// Unknown value tag inside a map.
|
||||
assert_eq!(
|
||||
decode_client(&[OP_SET_ROOM_PROPS, 1, 1, 0, b'A', 9]),
|
||||
Err(DecodeError::UnknownTag(9))
|
||||
);
|
||||
// A map that promises two entries but carries one.
|
||||
assert_eq!(
|
||||
decode_client(&[OP_SET_ROOM_PROPS, 2, 1, 0, b'A', TAG_NULL]),
|
||||
Err(DecodeError::Truncated)
|
||||
);
|
||||
// Invalid UTF-8 in a string.
|
||||
assert_eq!(decode_client(&[OP_JOIN_ROOM, 1, 0, 0xff]), Err(DecodeError::BadUtf8));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn duplicate_map_keys_collapse_last_writer_wins() {
|
||||
let bytes = vec![
|
||||
OP_SET_ROOM_PROPS, 2,
|
||||
1, 0, b'A', TAG_INT, 1, 0, 0, 0,
|
||||
1, 0, b'A', TAG_INT, 2, 0, 0, 0,
|
||||
];
|
||||
let ClientMsg::SetRoomProps { props } = decode_client(&bytes).unwrap() else {
|
||||
panic!("expected SetRoomProps");
|
||||
};
|
||||
assert_eq!(props.len(), 1);
|
||||
assert_eq!(props.get_int("A"), Some(2));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn map_set_is_last_writer_wins_in_place() {
|
||||
let mut map = Map::from_pairs(vec![("A", Value::Int(1)), ("B", Value::Int(2))]);
|
||||
map.set("A", Value::Int(9));
|
||||
assert_eq!(map.len(), 2);
|
||||
assert_eq!(
|
||||
map.iter().map(|(k, _)| k).collect::<Vec<_>>(),
|
||||
vec!["A", "B"]
|
||||
);
|
||||
assert_eq!(map.get_int("A"), Some(9));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bool_payload_accepts_any_nonzero() {
|
||||
// Photon writes 0/1; be lenient reading, strict writing.
|
||||
assert_eq!(
|
||||
decode_server(&[OP_PLAYER_LEFT, 1, 0, 0, 0, 7]),
|
||||
Ok(ServerMsg::PlayerLeft { actor: 1, inactive: true })
|
||||
);
|
||||
assert_eq!(
|
||||
encode_server(&ServerMsg::PlayerLeft { actor: 1, inactive: true })[5],
|
||||
1
|
||||
);
|
||||
}
|
||||
}
|
||||
2397
src/router/multi_live/rooms.rs
Normal file
2397
src/router/multi_live/rooms.rs
Normal file
File diff suppressed because it is too large
Load Diff
367
src/router/multi_live/ws.rs
Normal file
367
src/router/multi_live/ws.rs
Normal file
@@ -0,0 +1,367 @@
|
||||
// WebSocket end of the multi-live relay: GET /api/multi_live/ws.
|
||||
//
|
||||
// One task per connection reads frames, decodes them and drives the (synchronous,
|
||||
// lock-serialised) registry in rooms.rs. A second task per connection drains that
|
||||
// connection's unbounded queue into the socket. Nothing that touches the registry ever
|
||||
// awaits while the lock is held, which is what lets the registry hand out a total order
|
||||
// over every broadcast - see the header comment in rooms.rs.
|
||||
//
|
||||
// Everything the client can get wrong ends in a close: 4000 for a framing/protocol
|
||||
// error, 4001 for an unauthenticated or unauthenticated-first frame, 4002 for blowing
|
||||
// the inbound rate cap, 4003 for going silent long enough to be presumed dead (the
|
||||
// liveness sweep - see LIVENESS_TIMEOUT in rooms.rs, and start_sweeper below).
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use actix_web::rt;
|
||||
use actix_web::{web, HttpRequest, HttpResponse};
|
||||
use actix_ws::{AggregatedMessage, AggregatedMessageStream, CloseCode, CloseReason, Session};
|
||||
use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
|
||||
|
||||
use crate::router::userdata;
|
||||
use super::proto::{
|
||||
self, ClientMsg, DecodeError, ServerMsg, CLOSE_PROTOCOL, CLOSE_RATE_LIMIT,
|
||||
CLOSE_UNAUTHENTICATED,
|
||||
};
|
||||
use super::rooms::{self, ConnId, Outbound};
|
||||
|
||||
// "Server closes idle unauthenticated connections after 10s."
|
||||
const AUTH_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
// "a per-connection cap of 30 inbound messages/sec ... over the cap -> close 4002".
|
||||
// The honest client peaks at 3-5/sec.
|
||||
const RATE_LIMIT_PER_SEC: usize = 30;
|
||||
const RATE_WINDOW: Duration = Duration::from_secs(1);
|
||||
// Property bags and RPC params are tens of bytes; nothing legitimate comes close.
|
||||
const MAX_FRAME_BYTES: usize = 64 * 1024;
|
||||
// How often the inactive-actor, empty-room and connection-liveness expiries are checked.
|
||||
const SWEEP_INTERVAL: Duration = Duration::from_secs(1);
|
||||
|
||||
pub async fn ws(req: HttpRequest, body: web::Payload) -> Result<HttpResponse, actix_web::Error> {
|
||||
// Older client builds carry incompatible multi implementations (pre-relay wire
|
||||
// framing), so the upgrade itself is gated on the extended-protocol version — a
|
||||
// non-101 answer makes the client's ConnectAsync fail cleanly into its official
|
||||
// disconnect flow instead of mis-framing against the relay.
|
||||
let protocol = crate::router::global::client_protocol_version(&req);
|
||||
if protocol < super::PROTOCOL_VERSION {
|
||||
println!(
|
||||
"multi_live/ws: upgrade refused, X-Protocol-Version {} < {}",
|
||||
protocol,
|
||||
super::PROTOCOL_VERSION
|
||||
);
|
||||
return Ok(HttpResponse::UpgradeRequired().finish());
|
||||
}
|
||||
let (response, session, stream) = match actix_ws::handle(&req, body) {
|
||||
Ok(parts) => {
|
||||
println!("multi_live/ws: upgrade accepted, awaiting Auth");
|
||||
parts
|
||||
}
|
||||
Err(err) => {
|
||||
println!("multi_live/ws: upgrade REJECTED (not a websocket handshake?): {}", err);
|
||||
return Err(err);
|
||||
}
|
||||
};
|
||||
let stream = stream
|
||||
.max_frame_size(MAX_FRAME_BYTES)
|
||||
.aggregate_continuations()
|
||||
.max_continuation_size(MAX_FRAME_BYTES);
|
||||
|
||||
let (tx, rx) = unbounded_channel::<Outbound>();
|
||||
rt::spawn(writer(session, rx));
|
||||
rt::spawn(reader(stream, tx));
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
// The expiry timers: held-slot TTL, empty-room TTL and the connection liveness window.
|
||||
//
|
||||
// One task for the whole process, started from run_server so it lives on the system
|
||||
// arbiter. It used to be lazily started by the first WebSocket upgrade instead, which put
|
||||
// it on whichever HTTP worker happened to serve that upgrade: a panic anywhere in that
|
||||
// worker took the sweeper down with it, permanently and silently, and `Once` guaranteed
|
||||
// nothing would ever start it again. Every room in the process would then hold its seats
|
||||
// forever.
|
||||
//
|
||||
// Called once per run_server, unconditionally and not behind a "started already" flag:
|
||||
// each run_server builds its own actix System, and a task spawned on the previous one died
|
||||
// with it (the mobile path stops and restarts the server in-process). A flag would leave
|
||||
// the restarted server with no sweeper at all; two sweepers, if they ever overlapped,
|
||||
// would only mean the idempotent sweep runs twice a second.
|
||||
pub fn start_sweeper() {
|
||||
rt::spawn(async {
|
||||
let mut ticker = rt::time::interval(SWEEP_INTERVAL);
|
||||
loop {
|
||||
ticker.tick().await;
|
||||
// The guard is a temporary: it is released before the next await.
|
||||
rooms::registry().sweep(Instant::now());
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Drains this connection's FIFO queue to the socket. The queue is unbounded so that the
|
||||
// registry can push a whole broadcast while holding its lock without ever blocking.
|
||||
async fn writer(mut session: Session, mut rx: UnboundedReceiver<Outbound>) {
|
||||
while let Some(out) = rx.recv().await {
|
||||
match out {
|
||||
Outbound::Msg(msg) => {
|
||||
if session.binary(proto::encode_server(&msg)).await.is_err() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
Outbound::Pong(bytes) => {
|
||||
if session.pong(&bytes).await.is_err() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
Outbound::Close(code) => {
|
||||
let _ = session
|
||||
.close(Some(CloseReason { code: CloseCode::Other(code), description: None }))
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
// Queue closed: the connection was deregistered, so the socket goes with it.
|
||||
let _ = session.close(None).await;
|
||||
}
|
||||
|
||||
async fn reader(mut stream: AggregatedMessageStream, tx: UnboundedSender<Outbound>) {
|
||||
let Some(id) = authenticate(&mut stream, &tx).await else {
|
||||
return;
|
||||
};
|
||||
|
||||
let mut rate = RateLimiter::new();
|
||||
loop {
|
||||
let Some(frame) = stream.recv().await else {
|
||||
break;
|
||||
};
|
||||
let payload = match frame {
|
||||
Ok(AggregatedMessage::Binary(bytes)) => bytes,
|
||||
Ok(AggregatedMessage::Ping(bytes)) => {
|
||||
// WebSocket-level keepalive, unrelated to the protocol's Ping op. It is
|
||||
// still proof the socket is alive, so it counts for the liveness window
|
||||
// (the protocol ops are touched inside registry.handle).
|
||||
rooms::registry().touch(id, Instant::now());
|
||||
let _ = tx.send(Outbound::Pong(bytes.to_vec()));
|
||||
continue;
|
||||
}
|
||||
Ok(AggregatedMessage::Pong(_)) => {
|
||||
rooms::registry().touch(id, Instant::now());
|
||||
continue;
|
||||
}
|
||||
Ok(AggregatedMessage::Close(_)) => break,
|
||||
Ok(AggregatedMessage::Text(_)) => {
|
||||
// "binary frames only ... Text frames are a protocol error -> close".
|
||||
println!("multi_live/ws: text frame from conn {}", id);
|
||||
close(&tx, CLOSE_PROTOCOL);
|
||||
break;
|
||||
}
|
||||
Err(err) => {
|
||||
println!("multi_live/ws: websocket error on conn {}: {}", id, err);
|
||||
break;
|
||||
}
|
||||
};
|
||||
|
||||
let now = Instant::now();
|
||||
if !rate.allow(now) {
|
||||
println!("multi_live/ws: conn {} exceeded {} msg/sec", id, RATE_LIMIT_PER_SEC);
|
||||
close(&tx, CLOSE_RATE_LIMIT);
|
||||
break;
|
||||
}
|
||||
|
||||
let msg = match proto::decode_client(&payload) {
|
||||
Ok(msg) => msg,
|
||||
Err(err) => {
|
||||
println!("multi_live/ws: bad frame from conn {}: {}", id, err);
|
||||
close(&tx, close_code_for(&err));
|
||||
break;
|
||||
}
|
||||
};
|
||||
if matches!(msg, ClientMsg::Auth { .. }) {
|
||||
// Auth is the first message and only the first message.
|
||||
println!("multi_live/ws: repeat Auth on conn {}", id);
|
||||
close(&tx, CLOSE_PROTOCOL);
|
||||
break;
|
||||
}
|
||||
rooms::registry().handle(id, msg, now);
|
||||
}
|
||||
|
||||
rooms::registry().disconnect(id, Instant::now());
|
||||
}
|
||||
|
||||
// Reads the mandatory first frame and registers the connection. Returns None when the
|
||||
// connection was closed instead.
|
||||
async fn authenticate(
|
||||
stream: &mut AggregatedMessageStream,
|
||||
tx: &UnboundedSender<Outbound>,
|
||||
) -> Option<ConnId> {
|
||||
let first = match rt::time::timeout(AUTH_TIMEOUT, stream.recv()).await {
|
||||
Ok(Some(Ok(frame))) => frame,
|
||||
Ok(Some(Err(_))) | Ok(None) => return None,
|
||||
Err(_) => {
|
||||
println!("multi_live/ws: no Auth within {}s", AUTH_TIMEOUT.as_secs());
|
||||
close(tx, CLOSE_UNAUTHENTICATED);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
|
||||
let AggregatedMessage::Binary(payload) = first else {
|
||||
// "First message on a connection MUST be Auth; anything else -> close 4001."
|
||||
close(tx, CLOSE_UNAUTHENTICATED);
|
||||
return None;
|
||||
};
|
||||
let Ok(ClientMsg::Auth { user_id, token }) = proto::decode_client(&payload) else {
|
||||
println!("multi_live/ws: first message was not a decodable Auth frame");
|
||||
close(tx, CLOSE_UNAUTHENTICATED);
|
||||
return None;
|
||||
};
|
||||
|
||||
let Some(uid) = resolve_user(&user_id, &token) else {
|
||||
// Never print the token itself (it is the login credential); empty-vs-unknown is
|
||||
// the diagnostic that matters: empty means the client sent no credential at all
|
||||
// (the pre-fix Android builds sent UserSaveData.m_uuid, which device builds never
|
||||
// populate), unknown means a credential the tokens table has no row for.
|
||||
println!(
|
||||
"multi_live/ws: rejecting uid {}: bad session ({})",
|
||||
user_id,
|
||||
if token.is_empty() { "empty token" } else { "unknown token" }
|
||||
);
|
||||
close(tx, CLOSE_UNAUTHENTICATED);
|
||||
return None;
|
||||
};
|
||||
|
||||
let id = rooms::registry().connect(uid, tx.clone(), Instant::now());
|
||||
println!("multi_live/ws: auth ok, uid {} connected as conn {}", uid, id);
|
||||
let _ = tx.send(Outbound::Msg(ServerMsg::AuthOk {
|
||||
actorless_time_ms: rooms::server_time_ms(),
|
||||
}));
|
||||
Some(id)
|
||||
}
|
||||
|
||||
// The relay validates the same credential the HTTP layer does. Over HTTP the login token
|
||||
// arrives inside the `a6573cbe` header (global::get_login) and every handler then keys
|
||||
// userdata off it; here it arrives as the Auth payload instead, and the tokens table maps
|
||||
// it back to the account. The claimed userId only has to agree with the token, so a
|
||||
// client cannot relay as somebody else - which is a little stricter than the HTTP layer,
|
||||
// where an unknown token simply resolves to an empty account.
|
||||
fn resolve_user(user_id: &str, token: &str) -> Option<i64> {
|
||||
if token.is_empty() {
|
||||
return None;
|
||||
}
|
||||
match_claim(user_id, userdata::uid_from_login_token(token))
|
||||
}
|
||||
|
||||
// The claim check, split out from the lookup so it can be tested without a database.
|
||||
// `uid` is 0 when the token is unknown.
|
||||
fn match_claim(user_id: &str, uid: i64) -> Option<i64> {
|
||||
if uid == 0 {
|
||||
return None;
|
||||
}
|
||||
match user_id.parse::<i64>() {
|
||||
// A blank or unparsable userId is tolerated: the token is the authority.
|
||||
Err(_) => Some(uid),
|
||||
Ok(claimed) if claimed == 0 || claimed == uid => Some(uid),
|
||||
Ok(_) => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn close_code_for(err: &DecodeError) -> u16 {
|
||||
// The spec names 4001 and 4002 only; every framing failure shares 4000.
|
||||
match err {
|
||||
DecodeError::Empty
|
||||
| DecodeError::Truncated
|
||||
| DecodeError::TrailingBytes
|
||||
| DecodeError::BadUtf8
|
||||
| DecodeError::UnknownOp(_)
|
||||
| DecodeError::UnknownTag(_) => CLOSE_PROTOCOL,
|
||||
}
|
||||
}
|
||||
|
||||
fn close(tx: &UnboundedSender<Outbound>, code: u16) {
|
||||
let _ = tx.send(Outbound::Close(code));
|
||||
}
|
||||
|
||||
// Sliding one second window rather than a fixed bucket, so 30 messages either side of a
|
||||
// second boundary still trips the cap.
|
||||
struct RateLimiter {
|
||||
seen: VecDeque<Instant>,
|
||||
}
|
||||
|
||||
impl RateLimiter {
|
||||
fn new() -> Self {
|
||||
RateLimiter { seen: VecDeque::with_capacity(RATE_LIMIT_PER_SEC + 1) }
|
||||
}
|
||||
|
||||
fn allow(&mut self, now: Instant) -> bool {
|
||||
while let Some(front) = self.seen.front() {
|
||||
if now.duration_since(*front) >= RATE_WINDOW {
|
||||
self.seen.pop_front();
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
if self.seen.len() >= RATE_LIMIT_PER_SEC {
|
||||
return false;
|
||||
}
|
||||
self.seen.push_back(now);
|
||||
true
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn rate_limiter_allows_the_cap_and_rejects_the_next() {
|
||||
let mut rate = RateLimiter::new();
|
||||
let start = Instant::now();
|
||||
for i in 0..RATE_LIMIT_PER_SEC {
|
||||
assert!(rate.allow(start + Duration::from_millis(i as u64)), "message {} refused", i);
|
||||
}
|
||||
assert!(!rate.allow(start + Duration::from_millis(999)));
|
||||
// The window slides: once the first message ages out there is room again.
|
||||
assert!(rate.allow(start + Duration::from_millis(1001)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rate_limiter_catches_a_burst_straddling_a_second_boundary() {
|
||||
let mut rate = RateLimiter::new();
|
||||
let start = Instant::now();
|
||||
// 29 messages at the end of one second...
|
||||
for i in 0..RATE_LIMIT_PER_SEC - 1 {
|
||||
assert!(rate.allow(start + Duration::from_millis(900 + i as u64)));
|
||||
}
|
||||
// ...and two more just after the boundary is still 31 inside one second.
|
||||
assert!(rate.allow(start + Duration::from_millis(1000)));
|
||||
assert!(!rate.allow(start + Duration::from_millis(1001)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn every_decode_failure_closes_with_the_protocol_code() {
|
||||
for err in [
|
||||
DecodeError::Empty,
|
||||
DecodeError::Truncated,
|
||||
DecodeError::TrailingBytes,
|
||||
DecodeError::BadUtf8,
|
||||
DecodeError::UnknownOp(200),
|
||||
DecodeError::UnknownTag(9),
|
||||
] {
|
||||
assert_eq!(close_code_for(&err), CLOSE_PROTOCOL);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn auth_needs_a_token_that_agrees_with_the_claimed_user() {
|
||||
// An empty token never reaches the database.
|
||||
assert_eq!(resolve_user("1", ""), None);
|
||||
// An unknown token resolves to uid 0.
|
||||
assert_eq!(match_claim("1", 0), None);
|
||||
assert_eq!(match_claim("42", 42), Some(42));
|
||||
// Claiming somebody else's account with your own token is refused.
|
||||
assert_eq!(match_claim("43", 42), None);
|
||||
// A blank, zero or unparsable userId defers to the token.
|
||||
assert_eq!(match_claim("", 42), Some(42));
|
||||
assert_eq!(match_claim("0", 42), Some(42));
|
||||
assert_eq!(match_claim("not-a-number", 42), Some(42));
|
||||
}
|
||||
}
|
||||
@@ -246,6 +246,13 @@ fn get_uid(token: &str) -> i64 {
|
||||
data.parse::<i64>().unwrap_or(0)
|
||||
}
|
||||
|
||||
// The account a login token belongs to, 0 when the token is unknown. The HTTP layer
|
||||
// never needs this (it keys everything off the token itself), but the multi-live relay
|
||||
// authenticates a {userId, token} pair and has to check the two agree.
|
||||
pub fn uid_from_login_token(token: &str) -> i64 {
|
||||
get_uid(token)
|
||||
}
|
||||
|
||||
// Needed by gree
|
||||
pub fn get_login_token(uid: i64) -> String {
|
||||
let data = DATABASE.lock_and_select("SELECT token FROM tokens WHERE user_id=?1", params!(uid));
|
||||
@@ -475,6 +482,47 @@ pub fn save_acc_eventlogin(auth_key: &str, data: JsonValue) {
|
||||
pub fn save_server_data(auth_key: &str, data: JsonValue) {
|
||||
save_data(auth_key, "server_data", data);
|
||||
}
|
||||
|
||||
// Read-modify-write of one account's server_data as ONE atomic step.
|
||||
//
|
||||
// get_server_data + save_server_data open a connection each, so two requests for the same
|
||||
// account both read the pre-state and the later write wins - which for a record that is
|
||||
// meant to be spent exactly once (the started-live record /multi_live/end awards off) means
|
||||
// both ends see it and both award. Everything here happens inside a single BEGIN IMMEDIATE
|
||||
// transaction instead, so concurrent callers serialise and the second one observes what the
|
||||
// first one wrote.
|
||||
//
|
||||
// `f` sees the parsed server_data and returns whatever the caller needs out of it; the
|
||||
// (possibly mutated) value is written back before the transaction commits. get_key runs
|
||||
// BEFORE the transaction because it can create the account, which writes on its own
|
||||
// connection and would otherwise deadlock against our write lock.
|
||||
//
|
||||
// A database error yields T::default() rather than a panic: the callers all have a
|
||||
// "nothing to spend" branch, which is the right answer when the record could not be read.
|
||||
pub fn modify_server_data<T: Default>(auth_key: &str, f: impl FnOnce(&mut JsonValue) -> T) -> T {
|
||||
let key = get_key(auth_key);
|
||||
let rv = DATABASE.lock_and_transact(|conn| {
|
||||
let raw: String = conn.query_row(
|
||||
"SELECT server_data FROM server_data WHERE user_id=?1",
|
||||
params!(key),
|
||||
|row| row.get(0)
|
||||
)?;
|
||||
let mut data = jzon::parse(&raw).unwrap_or(JsonValue::Null);
|
||||
let rv = f(&mut data);
|
||||
conn.execute(
|
||||
"UPDATE server_data SET server_data=?1 WHERE user_id=?2",
|
||||
params!(jzon::stringify(data), key)
|
||||
)?;
|
||||
Ok(rv)
|
||||
});
|
||||
match rv {
|
||||
Ok(rv) => rv,
|
||||
Err(err) => {
|
||||
println!("modify_server_data: {} for user {}", err, key);
|
||||
T::default()
|
||||
}
|
||||
}
|
||||
}
|
||||
pub fn save_acc_chats(auth_key: &str, data: JsonValue) {
|
||||
save_data(auth_key, "chats", data);
|
||||
}
|
||||
|
||||
@@ -1,13 +1,29 @@
|
||||
{
|
||||
"point_ranking": {
|
||||
"rank": 0,
|
||||
"point": 0
|
||||
},
|
||||
"score_ranking": [],
|
||||
"member_ranking": [],
|
||||
"lottery_box": [],
|
||||
"score_ranking": {
|
||||
"all_rank": 0,
|
||||
"group_rank": 0,
|
||||
"score": 0
|
||||
},
|
||||
"member_ranking": {
|
||||
"master_character_id": 0,
|
||||
"rank": 0,
|
||||
"point": 0
|
||||
},
|
||||
"lottery_box": {
|
||||
"master_lottery_id": 0,
|
||||
"reset_count": 0,
|
||||
"draw_count_list": []
|
||||
},
|
||||
"mission_list": [],
|
||||
"policy_agreement": 0,
|
||||
"incentive_lottery": 0,
|
||||
"is_disconnected": 0,
|
||||
"help_count": 0,
|
||||
"penalty_remaining_time": 0,
|
||||
"star_event": {
|
||||
"star_level": 0,
|
||||
"last_event_star_level": 0,
|
||||
|
||||
24
src/sql.rs
24
src/sql.rs
@@ -59,4 +59,28 @@ impl SQLite {
|
||||
}
|
||||
Ok(rv)
|
||||
}
|
||||
|
||||
// Runs a read-modify-write as one unit. Additive on purpose — the other helpers open
|
||||
// a fresh connection per statement, so a caller that SELECTs then INSERTs races any
|
||||
// concurrent caller doing the same and the loser hits a constraint violation (which
|
||||
// lock_and_exec would unwrap into a worker panic).
|
||||
//
|
||||
// BEGIN IMMEDIATE takes the write lock up front rather than at first write, so two
|
||||
// callers serialise instead of both reading the pre-state; busy_timeout makes the
|
||||
// loser wait for the winner rather than fail instantly (SQLite::new sets that on its
|
||||
// own short-lived setup connection, not on the per-call ones).
|
||||
//
|
||||
// Errors are returned, never unwrapped: statistics writes must not take down a
|
||||
// request.
|
||||
pub fn lock_and_transact<T>(
|
||||
&self,
|
||||
f: impl FnOnce(&Connection) -> Result<T, rusqlite::Error>
|
||||
) -> Result<T, rusqlite::Error> {
|
||||
let mut conn = Connection::open(&self.path)?;
|
||||
conn.busy_timeout(std::time::Duration::from_secs(10))?;
|
||||
let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
|
||||
let rv = f(&tx)?;
|
||||
tx.commit()?;
|
||||
Ok(rv)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user