anna
/
megamappingway
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(server): switch to hashes for primary keys

This commit is contained in:
Anna 2023-10-07 04:33:15 -04:00
parent 03c67fa26c
commit 630b2ddc0b
Signed by: anna
GPG Key ID: D0943384CD9F87D1
4 changed files with 160 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
server/Cargo.lock generated
View File

@ -320,6 +320,12 @@ dependencies = [
"typenum",
]
[[package]]
name = "data-encoding"
version = "2.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2e66c9d817f1720209181c316d28635c050fa304f9c79e47a520882661b7308"
[[package]]
name = "der"
version = "0.7.8"
@ -1057,9 +1063,13 @@ dependencies = [
"axum",
"bytes",
"chrono",
"data-encoding",
"rand",
"rmp-serde",
"serde",
"serde_bytes",
"serde_path_to_error",
"siphasher",
"sqlx",
"tokio",
"tower",
@ -1219,6 +1229,15 @@ dependencies = [
"serde_derive",
]
[[package]]
name = "serde_bytes"
version = "0.11.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ab33ec92f677585af6d88c65593ae2375adde54efdbf16d597f2cbc7a6d368ff"
dependencies = [
"serde",
]
[[package]]
name = "serde_derive"
version = "1.0.188"
@ -1295,6 +1314,12 @@ dependencies = [
"rand_core",
]
[[package]]
name = "siphasher"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "54ac45299ccbd390721be55b412d41931911f654fa99e2cb8bfb57184b2061fe"
[[package]]
name = "slab"
version = "0.4.9"

4
server/Cargo.toml
View File

@ -10,9 +10,13 @@ anyhow = "1"
axum = "0.6"
bytes = "1"
chrono = { version = "0.4", features = ["serde"] }
data-encoding = "2"
rand = "0.8"
rmp-serde = "1"
serde = { version = "1", features = ["derive"] }
serde_bytes = "0.11"
serde_path_to_error = "0.1"
siphasher = "1"
sqlx = { version = "0.7", features = ["runtime-tokio", "sqlite", "chrono"] }
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
tower = "0.4"

6
server/migrations/1_initial_schema.up.sql
View File

@ -1,6 +1,6 @@
create table players
(
name text not null,
hash blob not null primary key,
world int not null,
timestamp timestamp with time zone not null,
territory int not null,
@ -14,9 +14,7 @@ create table players
job int not null,
free_company text,
current_hp int not null,
max_hp int not null,
primary key (name, world)
max_hp int not null
);
create index players_territory_idx on players (territory);

160
server/src/main.rs
View File

@ -1,5 +1,6 @@
use std::ops::{Deref, DerefMut};
use std::sync::Arc;
use std::time::Duration;
use anyhow::Result;
use axum::{async_trait, BoxError, Router, Server};
@ -13,8 +14,10 @@ use axum::http::Method;
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use bytes::Bytes;
use rand::distributions::{Alphanumeric, DistString};
use serde::{Deserialize, Serialize};
use serde::de::DeserializeOwned;
use siphasher::sip::SipHasher;
use sqlx::SqlitePool;
use tower::ServiceBuilder;
use tower_http::compression::CompressionLayer;
@ -23,11 +26,48 @@ use tower_http::decompression::RequestDecompressionLayer;
static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!();
struct AppState {
salt: String,
pool: Arc<SqlitePool>,
hasher: SipHasher,
}
#[tokio::main]
async fn main() -> Result<()> {
let pool = SqlitePool::connect("./database.sqlite").await?;
MIGRATOR.run(&pool).await?;
let pool = Arc::new(pool);
{
let pool = Arc::clone(&pool);
tokio::task::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(60 * 30));
loop {
interval.tick().await;
println!("deleting old records");
let result = sqlx::query!(
// language=sqlite
"delete from players where unixepoch('now') - unixepoch(timestamp) > 3600",
)
.execute(&*pool)
.await;
match result {
Ok(res) => println!("{} record(s) deleted", res.rows_affected()),
Err(e) => eprintln!("could not delete old records: {e:#}"),
}
}
});
}
let state = Arc::new(AppState {
pool,
salt: Alphanumeric.sample_string(&mut rand::thread_rng(), 8),
hasher: SipHasher::new(),
});
#[cfg(not(debug_assertions))]
let cors = CorsLayer::new()
.allow_origin([
@ -42,7 +82,7 @@ async fn main() -> Result<()> {
.route("/:territory", get(territory))
.route("/:territory/:world", get(territory_world))
.route("/upload", post(upload))
.with_state(Arc::new(pool))
.with_state(state)
.layer(cors)
.layer(CompressionLayer::new())
.layer(
@ -60,15 +100,16 @@ async fn main() -> Result<()> {
}
async fn territory(
State(pool): State<Arc<SqlitePool>>,
State(state): State<Arc<AppState>>,
Path(territory): Path<u32>,
) -> Result<MsgPack<Vec<AnonymousPlayerInfo>>, AppError>
{
let info = sqlx::query_as!(
AnonymousPlayerInfo,
AnonymousPlayerInfoInternal,
// language=sqlite
r#"
select world,
select hash,
world,
x,
y,
z,
@ -78,29 +119,35 @@ async fn territory(
job,
free_company,
current_hp,
max_hp
max_hp,
coalesce(unixepoch('now') - unixepoch(timestamp), 30) as age
from players
where territory = ?
and unixepoch('now') - unixepoch(timestamp) < 30
and age < 30
"#,
territory,
)
.fetch_all(&*pool)
.fetch_all(&*state.pool)
.await?;
let info = info.into_iter()
.map(|player| AnonymousPlayerInfo::new_from(player, &state.hasher, &state.salt, territory))
.collect();
Ok(MsgPack(info))
}
async fn territory_world(
State(pool): State<Arc<SqlitePool>>,
State(state): State<Arc<AppState>>,
Path((territory, world)): Path<(u32, u32)>,
) -> Result<MsgPack<Vec<AnonymousPlayerInfo>>, AppError>
{
let info = sqlx::query_as!(
AnonymousPlayerInfo,
AnonymousPlayerInfoInternal,
// language=sqlite
r#"
select world,
select hash,
world,
x,
y,
z,
@ -110,34 +157,37 @@ async fn territory_world(
job,
free_company,
current_hp,
max_hp
max_hp,
coalesce(unixepoch('now') - unixepoch(timestamp), 30) as age
from players
where territory = ?
and current_world = ?
and unixepoch('now') - unixepoch(timestamp) < 30
and age < 30
"#,
territory,
world,
)
.fetch_all(&*pool)
.fetch_all(&*state.pool)
.await?;
let info = info.into_iter()
.map(|player| AnonymousPlayerInfo::new_from(player, &state.hasher, &state.salt, territory))
.collect();
Ok(MsgPack(info))
}
async fn upload(
pool: State<Arc<SqlitePool>>,
state: State<Arc<AppState>>,
data: MsgPack<Update>,
) -> Result<(), AppError> {
let mut t = pool.begin().await?;
let mut t = state.pool.begin().await?;
for player in &data.players {
if !player.is_sane() {
continue;
}
let name = player.name.trim();
let fc = match &player.free_company {
Some(x) if x.trim() == "" => None,
Some(x) => Some(x),
@ -147,10 +197,10 @@ async fn upload(
sqlx::query!(
// language=sqlite
"
insert into players (name, world, timestamp, territory, current_world, x, y, z, w, customize, level, job, free_company, current_hp,
insert into players (hash, world, timestamp, territory, current_world, x, y, z, w, customize, level, job, free_company, current_hp,
max_hp)
values (?, ?, current_timestamp, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
on conflict (name, world) do update set timestamp = current_timestamp,
on conflict (hash) do update set timestamp = current_timestamp,
territory = ?,
current_world = ?,
x = ?,
@ -164,7 +214,7 @@ async fn upload(
current_hp = ?,
max_hp = ?
",
name,
player.hash,
player.world,
data.territory,
@ -210,12 +260,14 @@ struct Update {
#[derive(Deserialize)]
struct PlayerInfo {
name: String,
#[serde(with = "serde_bytes")]
hash: Vec<u8>,
world: u32,
x: f64,
y: f64,
z: f64,
w: f64,
#[serde(with = "serde_bytes")]
customize: Vec<u8>,
level: u8,
job: u32,
@ -226,32 +278,78 @@ struct PlayerInfo {
impl PlayerInfo {
fn is_sane(&self) -> bool {
if self.name.trim().is_empty() {
return false;
}
if self.world == 65535 || self.level == 0 {
return false;
}
true
!self.hash.is_empty()
&& self.world != 65535
&& self.level != 0
}
}
#[derive(Serialize)]
struct AnonymousPlayerInfo {
world: i64,
// timestamp: DateTime<Utc>,
x: f64,
y: f64,
z: f64,
w: f64,
#[serde(with = "serde_bytes")]
customize: Vec<u8>,
level: i64,
job: i64,
free_company: Option<String>,
current_hp: i64,
max_hp: i64,
age: i64,
territory_unique_id: u64,
}
impl AnonymousPlayerInfo {
fn new_from(value: AnonymousPlayerInfoInternal, hasher: &SipHasher, salt: &str, territory: u32) -> Self {
Self {
territory_unique_id: value.gen_hash(hasher, salt, territory),
world: value.world,
x: value.x,
y: value.y,
z: value.z,
w: value.w,
customize: value.customize,
level: value.level,
job: value.job,
free_company: value.free_company,
current_hp: value.current_hp,
max_hp: value.max_hp,
age: value.age,
}
}
}
#[derive(Serialize)]
struct AnonymousPlayerInfoInternal {
#[serde(skip)]
hash: Vec<u8>,
world: i64,
x: f64,
y: f64,
z: f64,
w: f64,
#[serde(with = "serde_bytes")]
customize: Vec<u8>,
level: i64,
job: i64,
free_company: Option<String>,
current_hp: i64,
max_hp: i64,
age: i64,
}
impl AnonymousPlayerInfoInternal {
pub fn gen_hash(&self, hasher: &SipHasher, salt: &str, territory: u32) -> u64 {
hasher.hash(format!(
"{}-{}-{}",
salt,
territory,
data_encoding::HEXLOWER.encode(&self.hash),
).as_bytes())
}
}
// Make our own error that wraps `anyhow::Error`.