GenarrativeAI/Genarrative
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: read gallery hot paths from spacetime cache

Browse Source
This commit is contained in:
kdletters
2026-05-17 00:03:07 +08:00
parent 99f539a601
commit d9c8473504
10 changed files with 347 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
server-rs/crates/spacetime-client/src/custom_world.rs
View File

@@ -181,6 +181,55 @@ impl SpacetimeClient {
pub async fn list_custom_world_gallery_entries(
&self,
) -> Result<Vec<CustomWorldGalleryEntryRecord>, SpacetimeClientError> {
let records = self.read_custom_world_gallery_entries_from_cache().await?;
if !records.is_empty()
|| self
.custom_world_gallery_legacy_sync_attempted
.swap(true, std::sync::atomic::Ordering::SeqCst)
{
return Ok(records);
}
let _ = self
.sync_custom_world_gallery_entries_via_legacy_procedure()
.await;
self.read_custom_world_gallery_entries_from_cache().await
}
async fn read_custom_world_gallery_entries_from_cache(
&self,
) -> Result<Vec<CustomWorldGalleryEntryRecord>, SpacetimeClientError> {
self.read_after_connect(move |connection| {
let recent_play_counts = public_work_recent_play_counts(connection, "custom-world");
let mut entries = connection
.db()
.custom_world_gallery_entry()
.iter()
.collect::<Vec<_>>();
entries.sort_by(|left, right| {
right
.published_at
.cmp(&left.published_at)
.then(right.updated_at.cmp(&left.updated_at))
});
Ok(entries
.into_iter()
.map(|entry| {
let recent_play_count_7d = recent_play_counts
.get(&entry.profile_id)
.copied()
.unwrap_or(0);
map_custom_world_gallery_entry_row(entry, recent_play_count_7d)
})
.collect())
})
.await
}
async fn sync_custom_world_gallery_entries_via_legacy_procedure(
&self,
) -> Result<(), SpacetimeClientError> {
self.call_after_connect(
"list_custom_world_gallery_entries",
move |connection, sender| {
@@ -188,8 +237,8 @@ impl SpacetimeClient {
.procedures()
.list_custom_world_gallery_entries_then(move |_, result| {
let mapped = result
.map_err(SpacetimeClientError::from_sdk_error)
.and_then(map_custom_world_gallery_list_result);
.map(|_| ())
.map_err(SpacetimeClientError::from_sdk_error);
send_once(&sender, mapped);
});
},

143
server-rs/crates/spacetime-client/src/lib.rs
View File

@@ -96,6 +96,7 @@ pub mod story_runtime;
pub mod visual_novel;
use std::{
collections::HashMap,
error::Error,
fmt,
sync::atomic::{AtomicBool, Ordering},
@@ -225,7 +226,7 @@ use module_story::{
use shared_kernel::format_timestamp_micros;
use spacetimedb_sdk::{DbContext, Table};
use tokio::{
sync::{OwnedSemaphorePermit, Semaphore, oneshot},
sync::{OwnedSemaphorePermit, RwLock, Semaphore, oneshot},
time::timeout,
};
@@ -257,6 +258,8 @@ pub struct AuthStoreSnapshotImportRecord {
pub struct SpacetimeClient {
config: SpacetimeClientConfig,
pool: Arc<SpacetimeConnectionPool>,
creation_entry_config_cache: Arc<RwLock<Option<CreationEntryConfigRecord>>>,
custom_world_gallery_legacy_sync_attempted: Arc<AtomicBool>,
}
#[derive(Debug)]
@@ -269,6 +272,8 @@ pub enum SpacetimeClientError {
}
const DEFAULT_PROCEDURE_TIMEOUT: Duration = Duration::from_secs(30);
const PUBLIC_WORK_PLAY_DAY_MICROS: i64 = 86_400_000_000;
const PUBLIC_WORK_RECENT_PLAY_WINDOW_DAYS: i64 = 7;
type ProcedureResultSender<T> =
Arc<Mutex<Option<oneshot::Sender<Result<T, SpacetimeClientError>>>>>;
@@ -286,7 +291,7 @@ struct PooledConnectionSlot {
struct PooledConnection {
connection: DbConnection,
_gallery_subscription: Vec<SubscriptionHandle>,
_read_model_subscriptions: Vec<SubscriptionHandle>,
runner: Option<JoinHandle<()>>,
broken: Arc<AtomicBool>,
}
@@ -321,7 +326,12 @@ impl SpacetimeClient {
permits: Arc::new(Semaphore::new(pool_size)),
});
Self { config, pool }
Self {
config,
pool,
creation_entry_config_cache: Arc::new(RwLock::new(None)),
custom_world_gallery_legacy_sync_attempted: Arc::new(AtomicBool::new(false)),
}
}
async fn call_after_connect<T>(
@@ -415,6 +425,14 @@ impl SpacetimeClient {
final_result
}
async fn cache_creation_entry_config(&self, config: CreationEntryConfigRecord) {
*self.creation_entry_config_cache.write().await = Some(config);
}
async fn read_cached_creation_entry_config(&self) -> Option<CreationEntryConfigRecord> {
self.creation_entry_config_cache.read().await.clone()
}
async fn acquire_connection(&self) -> Result<PooledConnectionLease, SpacetimeClientError> {
let permit = timeout(
self.config.procedure_timeout,
@@ -503,19 +521,19 @@ impl SpacetimeClient {
.map_err(|_| SpacetimeClientError::Timeout)?
.map_err(|_| SpacetimeClientError::ConnectDropped)??;
let gallery_subscription = self
.subscribe_puzzle_gallery_views(&connection, broken.clone())
let read_model_subscriptions = self
.subscribe_cached_read_models(&connection, broken.clone())
.await?;
Ok(PooledConnection {
connection,
_gallery_subscription: gallery_subscription,
_read_model_subscriptions: read_model_subscriptions,
runner: Some(runner),
broken,
})
}
async fn subscribe_puzzle_gallery_views(
async fn subscribe_cached_read_models(
&self,
connection: &DbConnection,
broken: Arc<AtomicBool>,
@@ -523,37 +541,67 @@ impl SpacetimeClient {
let mut subscriptions = Vec::new();
for query in [
"SELECT * FROM puzzle_gallery_view",
"SELECT * FROM custom_world_gallery_entry",
] {
let (sender, receiver) = oneshot::channel::<Result<(), SpacetimeClientError>>();
let applied_sender = Arc::new(Mutex::new(Some(sender)));
let on_applied_sender = applied_sender.clone();
let on_error_sender = applied_sender.clone();
let broken_flag = broken.clone();
let subscription = connection
.subscription_builder()
.on_applied(move |_| {
send_connect_once(&on_applied_sender, Ok(()));
})
.on_error(move |_, error| {
broken_flag.store(true, Ordering::SeqCst);
send_connect_once(
&on_error_sender,
Err(SpacetimeClientError::Procedure(error.to_string())),
);
})
.subscribe(query);
timeout(self.config.procedure_timeout, receiver)
.await
.map_err(|_| SpacetimeClientError::Timeout)?
.map_err(|_| SpacetimeClientError::ConnectDropped)??;
let subscription = self
.subscribe_cached_read_model_query(connection, broken.clone(), query, true)
.await?;
subscriptions.push(subscription);
}
for query in [
"SELECT * FROM public_work_play_daily_stat WHERE source_type = 'puzzle'",
"SELECT * FROM public_work_play_daily_stat WHERE source_type = 'custom-world'",
"SELECT * FROM creation_entry_config",
"SELECT * FROM creation_entry_type_config",
] {
if let Ok(subscription) = self
.subscribe_cached_read_model_query(connection, broken.clone(), query, false)
.await
{
subscriptions.push(subscription);
}
}
Ok(subscriptions)
}
async fn subscribe_cached_read_model_query(
&self,
connection: &DbConnection,
broken: Arc<AtomicBool>,
query: &'static str,
mark_broken_on_error: bool,
) -> Result<SubscriptionHandle, SpacetimeClientError> {
let (sender, receiver) = oneshot::channel::<Result<(), SpacetimeClientError>>();
let applied_sender = Arc::new(Mutex::new(Some(sender)));
let on_applied_sender = applied_sender.clone();
let on_error_sender = applied_sender.clone();
let broken_flag = broken.clone();
let subscription = connection
.subscription_builder()
.on_applied(move |_| {
send_connect_once(&on_applied_sender, Ok(()));
})
.on_error(move |_, error| {
if mark_broken_on_error {
broken_flag.store(true, Ordering::SeqCst);
}
send_connect_once(
&on_error_sender,
Err(SpacetimeClientError::Procedure(error.to_string())),
);
})
.subscribe(query);
timeout(self.config.procedure_timeout, receiver)
.await
.map_err(|_| SpacetimeClientError::Timeout)?
.map_err(|_| SpacetimeClientError::ConnectDropped)??;
Ok(subscription)
}
async fn release_connection(&self, mut lease: PooledConnectionLease) {
let mut slot_guard = self.pool.slots[lease.slot_index].lock().await;
slot_guard.in_use = false;
@@ -581,6 +629,39 @@ impl SpacetimeClient {
}
}
fn current_unix_micros() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|duration| duration.as_micros() as i64)
.unwrap_or(0)
}
fn current_public_work_day() -> i64 {
current_unix_micros().div_euclid(PUBLIC_WORK_PLAY_DAY_MICROS)
}
fn public_work_recent_play_counts(
connection: &DbConnection,
source_type: &str,
) -> HashMap<String, u32> {
let current_day = current_public_work_day();
let first_day = current_day - (PUBLIC_WORK_RECENT_PLAY_WINDOW_DAYS - 1);
let mut counts = HashMap::new();
for row in connection.db().public_work_play_daily_stat().iter() {
if row.source_type != source_type
|| row.played_day < first_day
|| row.played_day > current_day
{
continue;
}
let entry: &mut u32 = counts.entry(row.profile_id).or_insert(0);
*entry = (*entry).saturating_add(row.play_count);
}
counts
}
impl SpacetimeClientError {
pub(crate) fn from_sdk_error(error: impl fmt::Display) -> Self {
Self::Procedure(error.to_string())

88
server-rs/crates/spacetime-client/src/mapper.rs
View File

@@ -830,6 +830,48 @@ pub(crate) fn map_creation_entry_config_procedure_result(
))
}
pub(crate) fn build_creation_entry_config_record_from_rows(
header: CreationEntryConfig,
mut creation_types: Vec<CreationEntryTypeConfig>,
) -> CreationEntryConfigRecord {
creation_types.sort_by(|left, right| {
left.sort_order
.cmp(&right.sort_order)
.then_with(|| left.id.cmp(&right.id))
});
module_runtime::build_creation_entry_config_response(
module_runtime::CreationEntryConfigSnapshot {
config_id: header.config_id,
start_card: module_runtime::CreationEntryStartCardSnapshot {
title: header.start_title,
description: header.start_description,
idle_badge: header.start_idle_badge,
busy_badge: header.start_busy_badge,
},
type_modal: module_runtime::CreationEntryTypeModalSnapshot {
title: header.modal_title,
description: header.modal_description,
},
creation_types: creation_types
.into_iter()
.map(|item| module_runtime::CreationEntryTypeSnapshot {
id: item.id,
title: item.title,
subtitle: item.subtitle,
badge: item.badge,
image_src: item.image_src,
visible: item.visible,
open: item.open,
sort_order: item.sort_order,
updated_at_micros: item.updated_at.to_micros_since_unix_epoch(),
})
.collect(),
updated_at_micros: header.updated_at.to_micros_since_unix_epoch(),
},
)
}
fn map_creation_entry_config_snapshot(
snapshot: CreationEntryConfigSnapshot,
) -> module_runtime::CreationEntryConfigSnapshot {
@@ -1437,20 +1479,6 @@ pub(crate) fn map_custom_world_library_detail_result(
})
}
pub(crate) fn map_custom_world_gallery_list_result(
result: CustomWorldGalleryListResult,
) -> Result<Vec<CustomWorldGalleryEntryRecord>, SpacetimeClientError> {
if !result.ok {
return Err(SpacetimeClientError::procedure_failed(result.error_message));
}
Ok(result
.entries
.into_iter()
.map(map_custom_world_gallery_entry_snapshot)
.collect::<Result<Vec<_>, _>>()?)
}
pub(crate) fn map_custom_world_library_mutation_result(
result: CustomWorldLibraryMutationResult,
) -> Result<CustomWorldLibraryMutationRecord, SpacetimeClientError> {
@@ -2676,6 +2704,38 @@ pub(crate) fn map_custom_world_gallery_entry_snapshot(
})
}
pub(crate) fn map_custom_world_gallery_entry_row(
row: CustomWorldGalleryEntry,
recent_play_count_7d: u32,
) -> CustomWorldGalleryEntryRecord {
CustomWorldGalleryEntryRecord {
owner_user_id: row.owner_user_id,
profile_id: row.profile_id,
public_work_code: row.public_work_code,
author_public_user_code: row.author_public_user_code,
visibility: "published".to_string(),
published_at: Some(format_timestamp_micros(
row.published_at.to_micros_since_unix_epoch(),
)),
updated_at: format_timestamp_micros(row.updated_at.to_micros_since_unix_epoch()),
author_display_name: row.author_display_name,
world_name: row.world_name,
subtitle: row.subtitle,
summary_text: row.summary_text,
cover_image_src: row.cover_image_src,
theme_mode: format_custom_world_theme_mode(map_custom_world_theme_mode_back(
row.theme_mode,
))
.to_string(),
playable_npc_count: row.playable_npc_count,
landmark_count: row.landmark_count,
play_count: row.play_count,
remix_count: row.remix_count,
like_count: row.like_count,
recent_play_count_7d,
}
}
pub(crate) fn map_custom_world_published_profile_compile_snapshot(
snapshot: CustomWorldPublishedProfileCompileSnapshot,
) -> Result<CustomWorldPublishedProfileCompileRecord, SpacetimeClientError> {

47
server-rs/crates/spacetime-client/src/puzzle.rs
View File

@@ -5,45 +5,6 @@ use crate::module_bindings::delete_puzzle_work_procedure::delete_puzzle_work;
use crate::module_bindings::record_puzzle_work_like_procedure::record_puzzle_work_like;
use crate::module_bindings::remix_puzzle_work_procedure::remix_puzzle_work;
use crate::module_bindings::save_puzzle_ui_background_procedure::save_puzzle_ui_background;
use std::collections::HashMap;
use std::time::{SystemTime, UNIX_EPOCH};
const PUBLIC_WORK_PLAY_DAY_MICROS: i64 = 86_400_000_000;
const PUBLIC_WORK_RECENT_PLAY_WINDOW_DAYS: i64 = 7;
fn current_unix_micros() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_micros() as i64)
.unwrap_or(0)
}
fn current_public_work_day() -> i64 {
current_unix_micros().div_euclid(PUBLIC_WORK_PLAY_DAY_MICROS)
}
fn puzzle_gallery_recent_play_counts(connection: &DbConnection) -> HashMap<String, u32> {
let current_day = current_public_work_day();
let first_day = current_day - (PUBLIC_WORK_RECENT_PLAY_WINDOW_DAYS - 1);
let mut counts = HashMap::new();
for row in connection
.db()
.public_work_play_daily_stat()
.iter()
{
if row.source_type != "puzzle"
|| row.played_day < first_day
|| row.played_day > current_day
{
continue;
}
let entry: &mut u32 = counts.entry(row.profile_id).or_insert(0);
*entry = (*entry).saturating_add(row.play_count);
}
counts
}
impl SpacetimeClient {
pub async fn create_puzzle_agent_session(
@@ -443,9 +404,13 @@ impl SpacetimeClient {
&self,
) -> Result<Vec<PuzzleWorkProfileRecord>, SpacetimeClientError> {
self.read_after_connect(move |connection| {
let mut items = connection.db().puzzle_gallery_view().iter().collect::<Vec<_>>();
let mut items = connection
.db()
.puzzle_gallery_view()
.iter()
.collect::<Vec<_>>();
items.sort_by(|left, right| right.updated_at_micros.cmp(&left.updated_at_micros));
let recent_play_counts = puzzle_gallery_recent_play_counts(connection);
let recent_play_counts = public_work_recent_play_counts(connection, "puzzle");
Ok(items
.into_iter()
.map(|item| {

77
server-rs/crates/spacetime-client/src/runtime.rs
View File

@@ -3,6 +3,49 @@ use super::*;
impl SpacetimeClient {
pub async fn get_creation_entry_config(
&self,
) -> Result<CreationEntryConfigRecord, SpacetimeClientError> {
match self
.read_after_connect(move |connection| {
let config_id = module_runtime::CREATION_ENTRY_CONFIG_GLOBAL_ID.to_string();
let header = connection
.db()
.creation_entry_config()
.config_id()
.find(&config_id)
.ok_or_else(|| SpacetimeClientError::missing_snapshot("创作入口配置快照"))?;
let creation_types = connection
.db()
.creation_entry_type_config()
.iter()
.collect::<Vec<_>>();
Ok(build_creation_entry_config_record_from_rows(
header,
creation_types,
))
})
.await
{
Ok(config) => {
self.cache_creation_entry_config(config.clone()).await;
Ok(config)
}
Err(_) => {
if let Some(config) = self.read_cached_creation_entry_config().await {
return Ok(config);
}
match self.fetch_creation_entry_config_via_procedure().await {
Ok(config) => {
self.cache_creation_entry_config(config.clone()).await;
Ok(config)
}
Err(fallback_error) => Err(fallback_error),
}
}
}
}
async fn fetch_creation_entry_config_via_procedure(
&self,
) -> Result<CreationEntryConfigRecord, SpacetimeClientError> {
self.call_after_connect("get_creation_entry_config", move |connection, sender| {
connection
@@ -22,20 +65,26 @@ impl SpacetimeClient {
input: module_runtime::CreationEntryTypeAdminUpsertInput,
) -> Result<CreationEntryConfigRecord, SpacetimeClientError> {
let procedure_input: CreationEntryTypeAdminUpsertInput = input.into();
self.call_after_connect(
"upsert_creation_entry_type_config",
move |connection, sender| {
connection
.procedures()
.upsert_creation_entry_type_config_then(procedure_input, move |_, result| {
let mapped = result
.map_err(SpacetimeClientError::from_sdk_error)
.and_then(map_creation_entry_config_procedure_result);
send_once(&sender, mapped);
});
},
)
.await
let config = self
.call_after_connect(
"upsert_creation_entry_type_config",
move |connection, sender| {
connection
.procedures()
.upsert_creation_entry_type_config_then(
procedure_input,
move |_, result| {
let mapped = result
.map_err(SpacetimeClientError::from_sdk_error)
.and_then(map_creation_entry_config_procedure_result);
send_once(&sender, mapped);
},
);
},
)
.await?;
self.cache_creation_entry_config(config.clone()).await;
Ok(config)
}
pub async fn get_runtime_settings(

11
server-rs/crates/spacetime-client/src/telemetry.rs
View File

@@ -52,17 +52,10 @@ fn spacetime_metrics() -> &'static SpacetimeMetrics {
})
}
fn record_procedure(
procedure: &'static str,
duration: Duration,
failed: bool,
) {
fn record_procedure(procedure: &'static str, duration: Duration, failed: bool) {
let labels = vec![
KeyValue::new("procedure", procedure),
KeyValue::new(
"status_class",
if failed { "error" } else { "ok" },
),
KeyValue::new("status_class", if failed { "error" } else { "ok" }),
];
let metrics = spacetime_metrics();
metrics.calls.add(1, &labels);