diff --git a/.github/workflows/module-durable.yml b/.github/workflows/module-durable.yml index 889d6119b..36ac2a9ce 100644 --- a/.github/workflows/module-durable.yml +++ b/.github/workflows/module-durable.yml @@ -85,6 +85,7 @@ jobs: cargo test -p lyracore-gateway --test provision_cli provision_writes_the_account_to_the_world_shard_and_realm_core -- --ignored --exact cargo test -p lyracore-gateway --test startup_relays -- --ignored --test-threads=1 cargo test -p lyracore-gateway --test notification_events -- --ignored --test-threads=1 + cargo test -p lyracore-gateway --bin lyracore-gateway stdb::reads::indexed_durable_tests -- --ignored --test-threads=1 cargo test -p lyracore-gateway --bin lyracore-gateway stdb::world_view::item_durable_tests::live_worn_item_insert_enchant_and_delete_reach_a_nearby_peer -- --ignored --exact - name: Run Benilla against private Module instances diff --git a/docs/architecture.md b/docs/architecture.md index 6be6d41f3..1fa6e4bc3 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -596,9 +596,13 @@ The SDK cache has only unique-index finders, and a whole-table `iter()` copies e lock the pump needs. So relays and per-request reads use Gateway-side indexes, each kept from its table's insert, update and delete callbacks and registered before the subscription applies: `PartyMembershipIndex`, `ChannelIndex`, `GuildIndex`, `MailEscrowIndex`, `AuctionIndex`, -`ContactIndex` (owner to friends and ignores) and `CharacterNameIndex` (lowercase name to guid, -per Shard). Each lives on its connection generation, so a reconnect rebuilds it from the fresh -subscription. +`ContactIndex`, `CharacterNameIndex`, and indexes for inventory, skills, quest progress, loot +entitlement, target auras and enchantment effects. Each lives on its connection generation, so a +reconnect rebuilds it from the fresh subscription. Indexed inventory and projection reads hold a +snapshot read guard. The pump holds the write guard only while polling an SDK update, so these reads +see both the committed cache and all of that update's index callbacks. The guard releases while the +SDK awaits network input. Pump callbacks enqueue these guarded projections instead of reading them +synchronously. Combat, spell cast, spell impact, Speech and emote notifications use native `*_event_live` tables. The ordinary predecessors retain their published schemas for additive upgrades and expire through the diff --git a/gateway/src/codec/combat/subscriptions.rs b/gateway/src/codec/combat/subscriptions.rs index 5dde1fe94..6093e77e6 100644 --- a/gateway/src/codec/combat/subscriptions.rs +++ b/gateway/src/codec/combat/subscriptions.rs @@ -375,9 +375,8 @@ pub(crate) fn armor_packet( return None; } let guard = coord.0.coord(); - let db = &guard.conn.db; - let eff = crate::stdb::armor::effective_resistances(db, self_guid); - let pos = crate::stdb::armor::aura_armor_positive(db, self_guid); + let eff = crate::stdb::armor::effective_resistances(&guard, self_guid); + let pos = crate::stdb::armor::aura_armor_positive(&guard, self_guid); Some(Outbound::One(ServerOpcodeMessage::SMSG_UPDATE_OBJECT( Box::new(codec::build_armor_values(self_guid, eff, pos)), ))) diff --git a/gateway/src/codec/entity/subscriptions.rs b/gateway/src/codec/entity/subscriptions.rs index 227dfeedc..e63d456d3 100644 --- a/gateway/src/codec/entity/subscriptions.rs +++ b/gateway/src/codec/entity/subscriptions.rs @@ -11,6 +11,7 @@ use crate::stdb::views::entity_view; use crate::stdb::world_index::{CellKey, EntityLayer, SessionId, ShardId}; use crate::stdb::world_view::*; use crate::stdb::Coordinator; +use crate::stdb::LiveConn; use crate::world::Outbound; use anyhow::Result; use lyracore_shared::spatial::BOX_HALF_SPAN; @@ -39,7 +40,7 @@ pub(crate) fn build_peer_create( }; let view = peer_entity_view( row, - projected_dynamic_flags(&coord.0.coord().conn.db, viewer_guid, row), + projected_dynamic_flags(&coord.0.coord(), viewer_guid, row), |guid| guild_projection_of(coord, guid), ); let auras = world.auras.on_target(shard, row.guid); @@ -151,9 +152,9 @@ pub(crate) fn peer_create_outbound( struct LootTagProjectionRows<'a> { tap: Option<&'a CreatureQuestTap>, tag_group: Option<&'a CreatureLootTagGroup>, - tag_members: &'a [CreatureQuestTapMember], - current_members: &'a [GroupMember], - corpse_eligible: &'a [CorpseLootEligible], + tag_member: bool, + current_member: bool, + corpse_eligible: bool, } fn viewer_relative_dynamic_flags( @@ -164,11 +165,7 @@ fn viewer_relative_dynamic_flags( use lyracore_shared::constants::unit_dynamic_flags::{LOOTABLE, TAPPED_BY_PLAYER}; if entity.dead { - return if rows - .corpse_eligible - .iter() - .any(|row| row.corpse_guid == entity.guid && row.eligible_guid == viewer_guid) - { + return if rows.corpse_eligible { entity.dynamic_flags } else { entity.dynamic_flags & !LOOTABLE @@ -182,13 +179,7 @@ fn viewer_relative_dynamic_flags( .tag_group .filter(|group| group.creature_guid == entity.guid) { - Some(group) => { - rows.tag_members.iter().any(|member| { - member.creature_guid == entity.guid && member.character_guid == viewer_guid - }) && rows.current_members.iter().any(|member| { - member.character_guid == viewer_guid && member.group_id == group.group_id - }) - } + Some(_) => rows.tag_member && rows.current_member, None => viewer_guid == tap.character_guid, }; if entitled { @@ -199,44 +190,38 @@ fn viewer_relative_dynamic_flags( } pub(crate) fn projected_dynamic_flags( - db: &RemoteTables, + live: &LiveConn, viewer_guid: u64, entity: &WorldEntity, ) -> u32 { - let tap = db - .game_creature_quest_tap() - .creature_guid() - .find(&entity.guid); - let tag_group = db - .game_creature_loot_tag_group() - .creature_guid() - .find(&entity.guid); - let tag_members: Vec<_> = db - .game_creature_quest_tap_member() - .iter() - .filter(|member| member.creature_guid == entity.guid) - .collect(); - let current_members: Vec<_> = db - .game_group_member() - .iter() - .filter(|member| member.character_guid == viewer_guid) - .collect(); - let corpse_eligible: Vec<_> = db - .game_corpse_loot_eligible() - .iter() - .filter(|eligible| eligible.corpse_guid == entity.guid) - .collect(); - viewer_relative_dynamic_flags( - viewer_guid, - entity, - LootTagProjectionRows { - tap: tap.as_ref(), - tag_group: tag_group.as_ref(), - tag_members: &tag_members, - current_members: ¤t_members, - corpse_eligible: &corpse_eligible, - }, - ) + live.with_cache_snapshot(|| { + let db = &live.conn.db; + let tap = db + .game_creature_quest_tap() + .creature_guid() + .find(&entity.guid); + let tag_group = db + .game_creature_loot_tag_group() + .creature_guid() + .find(&entity.guid); + let loot_tags = live.loot_tags.read().unwrap(); + viewer_relative_dynamic_flags( + viewer_guid, + entity, + LootTagProjectionRows { + tap: tap.as_ref(), + tag_group: tag_group.as_ref(), + tag_member: loot_tags.is_tap_member(entity.guid, viewer_guid), + current_member: tag_group.as_ref().is_some_and(|group| { + live.party_memberships + .read() + .unwrap() + .contains(viewer_guid, group.group_id) + }), + corpse_eligible: loot_tags.is_corpse_eligible(entity.guid, viewer_guid), + }, + ) + }) } pub(crate) fn instance_relay_gate(row_instance_id: u64, viewer_instance_id: Option) -> bool { @@ -398,7 +383,7 @@ pub(crate) fn relay_entity_update( } let dynamic_flags = { let guard = coord.0.coord(); - projected_dynamic_flags(&guard.conn.db, viewer.self_guid, new) + projected_dynamic_flags(&guard, viewer.self_guid, new) }; let mut out: Vec = entity_update_to_outbound_with_dynamic_flags(old, new, dynamic_flags) @@ -479,7 +464,7 @@ pub(crate) fn loot_tag_flags_after_membership_change( Outbound::One(ServerOpcodeMessage::SMSG_UPDATE_OBJECT(Box::new( codec::build_dynamic_flags_values( entity.guid, - projected_dynamic_flags(&guard.conn.db, viewer.self_guid, &entity), + projected_dynamic_flags(&guard, viewer.self_guid, &entity), ), ))) }) @@ -963,9 +948,17 @@ pub(crate) mod tests { LootTagProjectionRows { tap, tag_group, - tag_members, - current_members, - corpse_eligible, + tag_member: tag_members.iter().any(|member| { + member.creature_guid == entity.guid && member.character_guid == viewer_guid + }), + current_member: tag_group.is_some_and(|group| { + current_members.iter().any(|member| { + member.character_guid == viewer_guid && member.group_id == group.group_id + }) + }), + corpse_eligible: corpse_eligible + .iter() + .any(|row| row.corpse_guid == entity.guid && row.eligible_guid == viewer_guid), }, ) } diff --git a/gateway/src/codec/item/subscriptions.rs b/gateway/src/codec/item/subscriptions.rs index 47e430cd8..c872d414b 100644 --- a/gateway/src/codec/item/subscriptions.rs +++ b/gateway/src/codec/item/subscriptions.rs @@ -4,10 +4,10 @@ use crate::stdb::reads::view_of_item_row; use crate::stdb::world_index::{EntityLayer, ShardId}; use crate::stdb::world_view::*; use crate::stdb::Coordinator; +use crate::stdb::LiveConn; use crate::world::Outbound; use lyracore_shared::item_property::client_enchantment_id; use lyracore_shared::spatial::BOX_HALF_SPAN; -use spacetimedb_sdk::Table; use std::sync::{Arc, OnceLock}; use wow_world_messages::vanilla::opcodes::ServerOpcodeMessage; @@ -111,7 +111,7 @@ pub(crate) fn trade_offer_extended( } pub(crate) fn item_gain_feedback( - db: &RemoteTables, + live: &LiveConn, self_guid: u64, slot: u8, entry: u32, @@ -133,14 +133,13 @@ pub(crate) fn item_gain_feedback( random_property_id, )), ))]; - let wanted = db.game_character_quest().iter().any(|q| { - q.character_guid == self_guid - && !q.rewarded + let wanted = live.quest_rows(self_guid).into_iter().any(|q| { + !q.rewarded && !q.failed - && db - .game_quest_objective() - .iter() - .any(|o| o.quest_entry == q.quest_entry && o.kind == 1 && o.target_entry == entry) + && live + .quest_objectives(q.quest_entry) + .into_iter() + .any(|o| o.kind == 1 && o.target_entry == entry) }); if wanted { use wow_world_messages::vanilla::SMSG_QUESTUPDATE_ADD_ITEM; @@ -155,11 +154,12 @@ pub(crate) fn item_gain_feedback( } pub(crate) fn item_instance_insert_outbound( - db: &RemoteTables, + live: &LiveConn, self_guid: u64, row: &ItemInstance, enchantment: u32, ) -> Vec { + let db = &live.conn.db; let mut out = Vec::new(); let (max_durability, container_slots) = db .game_item_template() @@ -194,10 +194,10 @@ pub(crate) fn item_instance_insert_outbound( ))); } if let Some((bag_slot, slot_in_bag)) = codec::bag_content_parts(row.slot) { - if let Some(bag) = db - .game_item_instance() - .iter() - .find(|item| item.owner_guid == self_guid && item.slot == bag_slot) + if let Some(bag) = live + .item_rows(self_guid) + .into_iter() + .find(|item| item.slot == bag_slot) { let (opcode, body) = codec::build_container_slot_values(bag.guid, slot_in_bag, row.guid); @@ -206,7 +206,7 @@ pub(crate) fn item_instance_insert_outbound( } // Queue CREATE and inventory pointers together before gain or quest-completion feedback. out.extend(item_gain_feedback( - db, + live, self_guid, row.slot, row.entry, @@ -215,13 +215,13 @@ pub(crate) fn item_instance_insert_outbound( row.random_property_id, )); if row.slot <= 18 { - append_item_armor_and_sheet(db, self_guid, &mut out); + append_item_armor_and_sheet(live, self_guid, &mut out); } out } pub(crate) fn item_instance_delete_outbound( - db: &RemoteTables, + live: &LiveConn, self_guid: u64, row: &ItemInstance, ) -> Vec { @@ -239,10 +239,10 @@ pub(crate) fn item_instance_delete_outbound( ))); } if let Some((bag_slot, slot_in_bag)) = codec::bag_content_parts(row.slot) { - if let Some(bag) = db - .game_item_instance() - .iter() - .find(|item| item.owner_guid == self_guid && item.slot == bag_slot) + if let Some(bag) = live + .item_rows(self_guid) + .into_iter() + .find(|item| item.slot == bag_slot) { let (opcode, body) = codec::build_container_slot_values(bag.guid, slot_in_bag, 0); out.push(Outbound::Raw { opcode, body }); @@ -252,25 +252,25 @@ pub(crate) fn item_instance_delete_outbound( codec::build_destroy_object(row.guid), ))); if row.slot <= 18 { - append_item_armor_and_sheet(db, self_guid, &mut out); + append_item_armor_and_sheet(live, self_guid, &mut out); } out } pub(crate) fn item_instance_update_outbound( - db: &RemoteTables, + live: &LiveConn, self_guid: u64, old: &ItemInstance, row: &ItemInstance, ) -> Vec { let old_slot_is_empty = old.slot != row.slot - && !db - .game_item_instance() - .iter() - .any(|item| item.owner_guid == self_guid && item.slot == old.slot); + && !live + .item_rows(self_guid) + .into_iter() + .any(|item| item.slot == old.slot); let mut out = if row.slot == old.slot && row.stack_count > old.stack_count { item_gain_feedback( - db, + live, self_guid, row.slot, row.entry, @@ -283,9 +283,9 @@ pub(crate) fn item_instance_update_outbound( }; if old.slot != row.slot { append_final_item_slots(self_guid, old.slot, row.slot, &mut out, |slot| { - db.game_item_instance() - .iter() - .find(|item| item.owner_guid == self_guid && item.slot == slot) + live.item_rows(self_guid) + .into_iter() + .find(|item| item.slot == slot) .map(|item| view_of_item_row(&item)) }); } @@ -306,7 +306,7 @@ pub(crate) fn item_instance_update_outbound( || row.slot <= 18 && ((old.durability == 0) != (row.durability == 0) || old.enchant_id != row.enchant_id) { - append_item_armor_and_sheet(db, self_guid, &mut out); + append_item_armor_and_sheet(live, self_guid, &mut out); } out } @@ -377,11 +377,12 @@ pub(crate) fn append_final_item_slots( } pub(crate) fn append_item_armor_and_sheet( - db: &RemoteTables, + live: &LiveConn, self_guid: u64, out: &mut Vec, ) { - let resistances = crate::stdb::armor::effective_resistances(db, self_guid); + let db = &live.conn.db; + let resistances = crate::stdb::armor::effective_resistances(live, self_guid); out.push(Outbound::One(ServerOpcodeMessage::SMSG_UPDATE_OBJECT( Box::new(codec::build_resistance_values(self_guid, resistances)), ))); @@ -416,7 +417,7 @@ pub(crate) fn item_inserted( let enchantment = item_enchantment_word(&coord, &row); let guard = coord.0.coord(); crate::codec::item::subscriptions::item_instance_insert_outbound( - &guard.conn.db, + &guard, viewer.self_guid, &row, enchantment, @@ -448,7 +449,7 @@ pub(crate) fn item_updated( item_owner_job(view, shard, row.owner_guid, move |viewer| { let guard = coord.0.coord(); crate::codec::item::subscriptions::item_instance_update_outbound( - &guard.conn.db, + &guard, viewer.self_guid, &old, &row, @@ -467,7 +468,7 @@ pub(crate) fn item_deleted( item_owner_job(view, shard, row.owner_guid, move |viewer| { let guard = coord.0.coord(); crate::codec::item::subscriptions::item_instance_delete_outbound( - &guard.conn.db, + &guard, viewer.self_guid, &row, ) @@ -483,11 +484,9 @@ pub(crate) fn item_slot_read(coord: &Coordinator) -> ItemSlotRead { coord .0 .coord() - .conn - .db - .game_item_instance() - .iter() - .find(|item| item.owner_guid == owner_guid && item.slot == slot) + .item_rows(owner_guid) + .into_iter() + .find(|item| item.slot == slot) .map(|item| crate::stdb::reads::view_of_item_row(&item)) }) } diff --git a/gateway/src/codec/quest/subscriptions.rs b/gateway/src/codec/quest/subscriptions.rs index 34c934ac4..713c8a213 100644 --- a/gateway/src/codec/quest/subscriptions.rs +++ b/gateway/src/codec/quest/subscriptions.rs @@ -3,34 +3,29 @@ use crate::stdb::bindings::*; use crate::stdb::world_index::ShardId; use crate::stdb::world_view::*; use crate::stdb::Coordinator; +use crate::stdb::LiveConn; use crate::world::Outbound; -use spacetimedb_sdk::Table; use wow_world_messages::vanilla::opcodes::ServerOpcodeMessage; -pub(crate) fn quest_log_outbound( - db: &crate::stdb::bindings::RemoteTables, - self_guid: u64, -) -> Option { +pub(crate) fn quest_log_outbound(live: &LiveConn, self_guid: u64) -> Option { if !crate::config::quest_log_fields_enabled() { return None; } - let quests: Vec<_> = db.game_character_quest().iter().collect(); - let slots = crate::stdb::reads::build_quest_log_slots(db, &quests, self_guid); + let slots = crate::stdb::reads::build_quest_log_slots(live, self_guid); let mask = codec::update_mask::full_quest_log_mask(&slots); let (opcode, body) = codec::build_values_update_raw(self_guid, &mask); Some(Outbound::Raw { opcode, body }) } pub(crate) fn quest_update_outbound( - db: &crate::stdb::bindings::RemoteTables, + live: &LiveConn, self_guid: u64, old: &CharacterQuest, row: &CharacterQuest, ) -> Vec { - let objectives: Vec<_> = db - .game_quest_objective() - .iter() - .filter(|objective| objective.quest_entry == row.quest_entry) + let objectives: Vec<_> = live + .quest_objectives(row.quest_entry) + .into_iter() .map(|objective| { ( objective.kind, @@ -40,7 +35,7 @@ pub(crate) fn quest_update_outbound( ) }) .collect(); - quest_update_packets(old, row, &objectives, quest_log_outbound(db, self_guid)) + quest_update_packets(old, row, &objectives, quest_log_outbound(live, self_guid)) } pub(crate) fn quest_update_feedback( @@ -105,7 +100,7 @@ pub(crate) fn quest_inserted( let (coord, self_guid) = (coord.clone(), viewer.self_guid); enqueue(viewer.clone(), move |_| { let guard = coord.0.coord(); - crate::codec::quest::subscriptions::quest_log_outbound(&guard.conn.db, self_guid) + crate::codec::quest::subscriptions::quest_log_outbound(&guard, self_guid) .into_iter() .collect() }); @@ -124,12 +119,7 @@ pub(crate) fn quest_updated( let (coord, self_guid, old, row) = (coord.clone(), viewer.self_guid, old.clone(), row.clone()); enqueue(viewer.clone(), move |_| { let guard = coord.0.coord(); - crate::codec::quest::subscriptions::quest_update_outbound( - &guard.conn.db, - self_guid, - &old, - &row, - ) + crate::codec::quest::subscriptions::quest_update_outbound(&guard, self_guid, &old, &row) }); } @@ -145,7 +135,7 @@ pub(crate) fn quest_deleted( let (coord, self_guid) = (coord.clone(), viewer.self_guid); enqueue(viewer.clone(), move |_| { let guard = coord.0.coord(); - crate::codec::quest::subscriptions::quest_log_outbound(&guard.conn.db, self_guid) + crate::codec::quest::subscriptions::quest_log_outbound(&guard, self_guid) .into_iter() .collect() }); @@ -182,10 +172,7 @@ fn quest_credit_granted( let (coord, quest_entry) = (coord.clone(), row.quest_entry); enqueue(viewer, move |viewer| { let guard = coord.0.coord(); - event_credit_packets( - quest_entry, - quest_log_outbound(&guard.conn.db, viewer.self_guid), - ) + event_credit_packets(quest_entry, quest_log_outbound(&guard, viewer.self_guid)) }); } @@ -211,7 +198,7 @@ fn quest_credit_changed( let coord = coord.clone(); enqueue(viewer, move |viewer| { let guard = coord.0.coord(); - quest_log_outbound(&guard.conn.db, viewer.self_guid) + quest_log_outbound(&guard, viewer.self_guid) .into_iter() .collect() }); @@ -222,7 +209,7 @@ fn quest_requirements_changed(view: &WorldView, coord: &Coordinator, shard: Shar let coord = coord.clone(); enqueue(viewer, move |viewer| { let guard = coord.0.coord(); - quest_log_outbound(&guard.conn.db, viewer.self_guid) + quest_log_outbound(&guard, viewer.self_guid) .into_iter() .collect() }); diff --git a/gateway/src/stdb/armor.rs b/gateway/src/stdb/armor.rs index a1272912f..13ad2e6ce 100644 --- a/gateway/src/stdb/armor.rs +++ b/gateway/src/stdb/armor.rs @@ -1,16 +1,17 @@ use super::bindings::*; -use spacetimedb_sdk::Table; -use std::collections::HashMap; +use super::connection::LiveConn; use lyracore_shared::{ aura::AuraAmount, resistance::{self, ResistanceItem}, }; -fn aura_values(db: &RemoteTables, guid: u64) -> Vec { - db.game_aura() - .iter() - .filter(move |a| a.target_guid == guid) +fn aura_values(live: &LiveConn, guid: u64) -> Vec { + live.combat_reads + .read() + .unwrap() + .auras(&live.conn.db, guid) + .into_iter() .map(|a| AuraAmount { kind: a.eff_kind, parameter: a.eff_p_0, @@ -46,81 +47,83 @@ pub(crate) fn sheet_stats(db: &RemoteTables, guid: u64) -> Option u32 { - resistance::positive_armor(aura_values(db, guid)) +pub(crate) fn aura_armor_positive(live: &LiveConn, guid: u64) -> u32 { + live.with_cache_snapshot(|| resistance::positive_armor(aura_values(live, guid))) } /// Descriptor order is armor, Holy, Fire, Nature, Frost, Shadow, Arcane. -pub(crate) fn effective_resistances(db: &RemoteTables, guid: u64) -> [u32; 7] { - let base_armor = db - .game_world_entity() - .guid() - .find(&guid) - .map_or(0, |e| e.armor); - let auras = aura_values(db, guid); - let items: Vec<_> = db - .game_item_instance() - .iter() - .filter(|i| i.owner_guid == guid) - .filter_map(|item| { - let tmpl = db.game_item_template().entry().find(&item.entry)?; - let property = (item.random_property_id != 0) - .then(|| { - db.game_item_random_property() - .property_id() - .find(&item.random_property_id) +pub(crate) fn effective_resistances(live: &LiveConn, guid: u64) -> [u32; 7] { + live.with_cache_snapshot(|| { + let db = &live.conn.db; + let base_armor = db + .game_world_entity() + .guid() + .find(&guid) + .map_or(0, |e| e.armor); + let auras = aura_values(live, guid); + let items: Vec<_> = live + .inventory + .read() + .unwrap() + .items(db, guid) + .into_iter() + .filter_map(|item| { + let tmpl = db.game_item_template().entry().find(&item.entry)?; + let property = (item.random_property_id != 0) + .then(|| { + db.game_item_random_property() + .property_id() + .find(&item.random_property_id) + }) + .flatten(); + Some(ResistanceItem { + slot: item.slot, + max_durability: tmpl.max_durability, + durability: item.durability, + template: [ + tmpl.stat_armor, + tmpl.holy_res, + tmpl.fire_res, + tmpl.nature_res, + tmpl.frost_res, + tmpl.shadow_res, + tmpl.arcane_res, + ], + enchant_id: item.enchant_id, + property_enchants: property + .map_or([0; 3], |p| [p.enchant_id_1, p.enchant_id_2, p.enchant_id_3]), }) - .flatten(); - Some(ResistanceItem { - slot: item.slot, - max_durability: tmpl.max_durability, - durability: item.durability, - template: [ - tmpl.stat_armor, - tmpl.holy_res, - tmpl.fire_res, - tmpl.nature_res, - tmpl.frost_res, - tmpl.shadow_res, - tmpl.arcane_res, - ], - enchant_id: item.enchant_id, - property_enchants: property - .map_or([0; 3], |p| [p.enchant_id_1, p.enchant_id_2, p.enchant_id_3]), }) + .collect(); + std::array::from_fn(|index| { + let bonus = resistance::resistance_bonus( + resistance::SCHOOL_BITS[index], + auras.iter().copied(), + items.iter().copied(), + |id, kind| { + lyracore_shared::item_property::enchant_stat( + kind, + u32::MAX, + live.combat_reads + .read() + .unwrap() + .enchantments(db, id) + .into_iter() + .map(|effect| (effect.kind, effect.amount, effect.school_mask)), + ) + }, + ); + resistance::effective_resistance(if index == 0 { base_armor } else { 0 }, bonus) }) - .collect(); - let mut enchantments = HashMap::>::new(); - for effect in db.game_item_enchantment().iter() { - enchantments.entry(effect.enchant_id).or_default().push(( - effect.kind, - effect.amount, - effect.school_mask, - )); - } - std::array::from_fn(|index| { - let bonus = resistance::resistance_bonus( - resistance::SCHOOL_BITS[index], - auras.iter().copied(), - items.iter().copied(), - |id, kind| { - lyracore_shared::item_property::enchant_stat( - kind, - u32::MAX, - enchantments.get(&id).into_iter().flatten().copied(), - ) - }, - ); - resistance::effective_resistance(if index == 0 { base_armor } else { 0 }, bonus) }) } -pub(crate) fn effective_armor(db: &RemoteTables, guid: u64) -> u32 { - effective_resistances(db, guid)[0] +pub(crate) fn effective_armor(live: &LiveConn, guid: u64) -> u32 { + effective_resistances(live, guid)[0] } -pub(crate) fn effective_magic_resistances(db: &RemoteTables, guid: u64) -> [u32; 6] { - effective_resistances(db, guid)[1..] +pub(crate) fn effective_magic_resistances(live: &LiveConn, guid: u64) -> [u32; 6] { + effective_resistances(live, guid)[1..] .try_into() .expect("the magic resistance slice has six schools") } @@ -180,7 +183,7 @@ mod tests { shard.assert_sql("UPDATE game_aura SET eff_kind = 161, eff_p0 = 1, amount = 100, stacks = 3 WHERE target_guid = 1"); assert!(poll_until(POLL_TIMEOUT, || { let guard = coordinator.0.coord(); - aura_armor_positive(&guard.conn.db, 1) == 900 + aura_armor_positive(&guard, 1) == 900 })); } } diff --git a/gateway/src/stdb/connection.rs b/gateway/src/stdb/connection.rs index ba4cc8141..dbde1f7f8 100644 --- a/gateway/src/stdb/connection.rs +++ b/gateway/src/stdb/connection.rs @@ -49,6 +49,11 @@ pub(crate) struct LiveConn { pub(crate) guilds: Arc>, pub(crate) contacts: Arc>, pub(crate) character_names: Arc>, + pub(crate) inventory: Arc>, + pub(crate) quests: Arc>, + pub(crate) loot_tags: Arc>, + pub(crate) combat_reads: Arc>, + cache_snapshot: Arc>, _sub: SubscriptionHandle, } @@ -95,6 +100,12 @@ impl PartyMembershipIndex { .copied() } + pub(crate) fn contains(&self, character_guid: u64, group_id: u64) -> bool { + self.by_character + .get(&character_guid) + .is_some_and(|rows| rows.values().any(|group| *group == group_id)) + } + pub(crate) fn member_row_ids(&self, group_id: u64) -> Vec { self.by_group .get(&group_id) @@ -213,10 +224,63 @@ async fn pump_messages( + snapshot: &RwLock<()>, + next: impl std::future::Future, +) -> T { + let mut next = std::pin::pin!(next); + std::future::poll_fn(|context| { + let _updating = snapshot.write().unwrap(); + std::future::Future::poll(next.as_mut(), context) + }) + .await +} + +#[cfg(test)] +mod cache_snapshot_tests { + use super::advance_cache_snapshot; + use std::future::Future; + use std::sync::RwLock; + use std::task::Poll; + + #[tokio::test] + async fn awaiting_a_message_leaves_the_cache_readable() { + let snapshot = RwLock::new(()); + let mut next = std::pin::pin!(advance_cache_snapshot( + &snapshot, + std::future::pending::<()>(), + )); + std::future::poll_fn(|context| { + assert!(next.as_mut().poll(context).is_pending()); + assert!(snapshot.try_read().is_ok()); + Poll::Ready(()) + }) + .await; + } + + #[tokio::test] + async fn an_applied_cache_is_unreadable_until_all_index_callbacks_finish() { + let snapshot = RwLock::new(()); + let cache = std::sync::atomic::AtomicU64::new(0); + let index = std::sync::atomic::AtomicU64::new(0); + let order = std::sync::atomic::Ordering::Relaxed; + advance_cache_snapshot(&snapshot, async { + cache.store(7, order); + assert!(snapshot.try_read().is_err()); + index.store(7, order); + }) + .await; + let _read = snapshot.read().unwrap(); + assert_eq!(cache.load(order), 7); + assert_eq!(index.load(order), 7); + } +} + // SDK 2.7.1 drains pending registrations before the next message, preventing a snapshot gap. fn start_pump( conn: &Arc, role: &str, + cache_snapshot: Arc>, ) -> Result<( std::thread::JoinHandle<()>, tokio::sync::mpsc::UnboundedSender, @@ -231,7 +295,7 @@ fn start_pump( .name("stdb-pump".into()) .spawn(move || { let result = runtime.block_on(pump_messages(conn.as_ref(), rx, || { - conn.advance_one_message_async() + advance_cache_snapshot(&cache_snapshot, conn.advance_one_message_async()) })); if let Err(error) = result { if !matches!(error, spacetimedb_sdk::Error::Disconnected) { @@ -243,6 +307,12 @@ fn start_pump( } impl LiveConn { + // Store reads and queued Relay jobs only; SDK row callbacks hold the snapshot lock. + pub(crate) fn with_cache_snapshot(&self, read: impl FnOnce() -> T) -> T { + let _snapshot = self.cache_snapshot.read().unwrap(); + read() + } + pub(super) fn is_healthy(&self) -> bool { self.conn.is_active() && self._sub.is_active() @@ -709,7 +779,8 @@ fn connect_subscribed( .map_err(|error| anyhow!("{role_label} build/connect failed: {error}"))?; let conn = Arc::new(conn); - let (pump, pump_commands) = start_pump(&conn, &role_label)?; + let cache_snapshot = Arc::new(RwLock::new(())); + let (pump, pump_commands) = start_pump(&conn, &role_label, cache_snapshot.clone())?; let party_memberships = Arc::new(RwLock::new(PartyMembershipIndex::default())); let inserted_memberships = party_memberships.clone(); conn.db.game_group_member().on_insert(move |_ctx, row| { @@ -733,6 +804,10 @@ fn connect_subscribed( let guilds = super::reads::watch_guilds(&conn); let contacts = super::reads::watch_contacts(&conn); let character_names = super::reads::watch_character_names(&conn); + let inventory = super::reads::watch_inventory(&conn); + let quests = super::reads::watch_quest(&conn); + let loot_tags = super::reads::watch_loot_tag(&conn); + let combat_reads = super::reads::watch_combat(&conn); let (tx, rx) = std::sync::mpsc::channel::>(); let tx_err = tx.clone(); let applied_commands = pump_commands.clone(); @@ -787,6 +862,11 @@ fn connect_subscribed( guilds, contacts, character_names, + inventory, + quests, + loot_tags, + combat_reads, + cache_snapshot, _sub: sub, }) } diff --git a/gateway/src/stdb/mod.rs b/gateway/src/stdb/mod.rs index c471b8d99..4845046c2 100644 --- a/gateway/src/stdb/mod.rs +++ b/gateway/src/stdb/mod.rs @@ -18,7 +18,7 @@ pub(crate) mod world_view; pub use connection::Coordinator; pub(crate) use connection::ReducerCallError; -pub(crate) use connection::{classify, DurableFailure}; +pub(crate) use connection::{classify, DurableFailure, LiveConn}; pub use subscriptions::PlayerSubscriptions; #[allow(unused_imports)] pub use views::{AccountRow, RealmRow}; diff --git a/gateway/src/stdb/reads/channel.rs b/gateway/src/stdb/reads/channel.rs index 809d68127..ef2828dda 100644 --- a/gateway/src/stdb/reads/channel.rs +++ b/gateway/src/stdb/reads/channel.rs @@ -1,6 +1,6 @@ use std::collections::{BTreeMap, HashMap}; -use super::*; +use crate::stdb::bindings::{ChatChannel, ChatChannelMember}; #[derive(Default)] pub(crate) struct ChannelIndex { diff --git a/gateway/src/stdb/reads/combat_index.rs b/gateway/src/stdb/reads/combat_index.rs new file mode 100644 index 000000000..59abbdb50 --- /dev/null +++ b/gateway/src/stdb/reads/combat_index.rs @@ -0,0 +1,169 @@ +use std::collections::{BTreeSet, HashMap}; +use std::sync::{Arc, RwLock}; + +use spacetimedb_sdk::{Table, TableWithPrimaryKey}; + +use super::super::bindings::*; + +#[derive(Default)] +pub(crate) struct CombatIndex { + auras: HashMap>, + enchantments: HashMap>, +} + +impl CombatIndex { + fn insert_auras(&mut self, row: &Aura) { + self.auras + .entry(row.target_guid) + .or_default() + .insert(row.id); + } + + fn remove_auras(&mut self, row: &Aura) { + if let Some(ids) = self.auras.get_mut(&row.target_guid) { + ids.remove(&row.id); + if ids.is_empty() { + self.auras.remove(&row.target_guid); + } + } + } + + pub(crate) fn aura_ids(&self, target_guid: u64) -> Vec { + self.auras + .get(&target_guid) + .into_iter() + .flatten() + .copied() + .collect() + } + + pub(crate) fn auras(&self, db: &RemoteTables, target_guid: u64) -> Vec { + self.aura_ids(target_guid) + .into_iter() + .filter_map(|id| db.game_aura().id().find(&id)) + .filter(|row| row.target_guid == target_guid) + .collect() + } + + fn insert_enchantments(&mut self, row: &ItemEnchantment) { + self.enchantments + .entry(row.enchant_id) + .or_default() + .insert(row.id); + } + + fn remove_enchantments(&mut self, row: &ItemEnchantment) { + if let Some(ids) = self.enchantments.get_mut(&row.enchant_id) { + ids.remove(&row.id); + if ids.is_empty() { + self.enchantments.remove(&row.enchant_id); + } + } + } + + pub(crate) fn enchantment_ids(&self, enchant_id: u32) -> Vec { + self.enchantments + .get(&enchant_id) + .into_iter() + .flatten() + .copied() + .collect() + } + + pub(crate) fn enchantments(&self, db: &RemoteTables, enchant_id: u32) -> Vec { + self.enchantment_ids(enchant_id) + .into_iter() + .filter_map(|id| db.game_item_enchantment().id().find(&id)) + .filter(|row| row.enchant_id == enchant_id) + .collect() + } +} + +pub(crate) fn watch_combat(conn: &DbConnection) -> Arc> { + let index = Arc::new(RwLock::new(CombatIndex::default())); + let inserted = index.clone(); + conn.db.game_aura().on_insert(move |_ctx, row| { + inserted.write().unwrap().insert_auras(row); + }); + let updated = index.clone(); + conn.db.game_aura().on_update(move |_ctx, old, new| { + let mut index = updated.write().unwrap(); + index.remove_auras(old); + index.insert_auras(new); + }); + let deleted = index.clone(); + conn.db.game_aura().on_delete(move |_ctx, row| { + deleted.write().unwrap().remove_auras(row); + }); + let inserted = index.clone(); + conn.db.game_item_enchantment().on_insert(move |_ctx, row| { + inserted.write().unwrap().insert_enchantments(row); + }); + let updated = index.clone(); + conn.db + .game_item_enchantment() + .on_update(move |_ctx, old, new| { + let mut index = updated.write().unwrap(); + index.remove_enchantments(old); + index.insert_enchantments(new); + }); + let deleted = index.clone(); + conn.db.game_item_enchantment().on_delete(move |_ctx, row| { + deleted.write().unwrap().remove_enchantments(row); + }); + index +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::codec::combat::subscriptions::tests::aura; + + #[test] + fn aura_retargeting_and_removal_leave_each_target_its_current_rows() { + let mut index = CombatIndex::default(); + let old = aura(7, 0, 0, 0, 0, 0); + index.insert_auras(&old); + let mut other = aura(8, 0, 0, 0, 0, 0); + other.id = 2; + index.insert_auras(&other); + assert_eq!(index.aura_ids(7), [1]); + let mut moved = old.clone(); + moved.target_guid = 8; + index.remove_auras(&old); + index.insert_auras(&moved); + assert!(index.aura_ids(7).is_empty()); + assert_eq!(index.aura_ids(8), [1, 2]); + index.remove_auras(&moved); + assert_eq!(index.aura_ids(8), [2]); + } + + #[test] + fn enchantment_effects_follow_their_enchantment_on_update_and_delete() { + let mut index = CombatIndex::default(); + let old = ItemEnchantment { + id: 1, + enchant_id: 10, + effect_index: 0, + kind: 8, + amount: 3, + spell_id: 0, + school_mask: 2, + }; + index.insert_enchantments(&old); + index.insert_enchantments(&ItemEnchantment { + id: 2, + ..old.clone() + }); + let moved = ItemEnchantment { + enchant_id: 20, + ..old.clone() + }; + index.remove_enchantments(&old); + index.insert_enchantments(&moved); + assert_eq!(index.enchantment_ids(10), [2]); + assert_eq!(index.enchantment_ids(20), [1]); + index.remove_enchantments(&moved); + assert!(index.enchantment_ids(20).is_empty()); + } +} diff --git a/gateway/src/stdb/reads/indexed_durable_tests.rs b/gateway/src/stdb/reads/indexed_durable_tests.rs new file mode 100644 index 000000000..ee43eab2e --- /dev/null +++ b/gateway/src/stdb/reads/indexed_durable_tests.rs @@ -0,0 +1,321 @@ +use std::sync::{mpsc, Arc, Mutex}; +use std::time::Duration; + +use spacetimedb_sdk::{DbContext, Table}; + +use crate::accept::BlockingTaskCapacity; +use crate::config::GatewayConfig; +use crate::durable_test_support::{poll_until, Standalone, POLL_TIMEOUT}; +use crate::stdb::bindings::*; +use crate::stdb::Coordinator; +use crate::world::QuestActionStore; + +struct SingleShardEnv { + previous: [Option; 3], + _guard: std::sync::MutexGuard<'static, ()>, +} + +const TOPOLOGY_VARS: [&str; 3] = [ + "LYRACORE_SHARD_MAP", + "LYRACORE_SHARD_MAP_FILE", + "LYRACORE_REALM_CORE", +]; + +impl SingleShardEnv { + fn install() -> Self { + let guard = crate::stdb::subscriptions::DURABLE_TOPOLOGY_ENV_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let previous = TOPOLOGY_VARS.map(std::env::var_os); + for name in TOPOLOGY_VARS { + std::env::remove_var(name); + } + Self { + previous, + _guard: guard, + } + } +} + +impl Drop for SingleShardEnv { + fn drop(&mut self) { + for (name, previous) in TOPOLOGY_VARS.into_iter().zip(&self.previous) { + match previous { + Some(value) => std::env::set_var(name, value), + None => std::env::remove_var(name), + } + } + } +} + +fn connect(shard: &Standalone, runtime: &tokio::runtime::Runtime) -> Coordinator { + let cfg = GatewayConfig { + logon_bind: "127.0.0.1:0".into(), + world_bind: "127.0.0.1:0".into(), + stdb_uri: shard.server().into(), + module_name: shard.shard_name().into(), + coordinator_token: Some(shard.owner_token()), + gateway_id: "indexed-durable-test".into(), + blocking_task_capacity: BlockingTaskCapacity::new(1), + }; + runtime.block_on(Coordinator::connect(&cfg)).unwrap() +} + +fn fence(coordinator: &Coordinator) -> u64 { + coordinator + .0 + .coord() + .durable_character_presence_snapshot(1, coordinator.shard_name()) + .unwrap() + .generation +} + +fn items(coordinator: &Coordinator, guid: u64) -> Vec<(u64, u32, u8, u32)> { + coordinator + .player_items(guid) + .unwrap() + .into_iter() + .map(|item| { + assert_eq!(item.owner_guid, guid); + (item.guid, item.entry, item.slot, item.stack_count) + }) + .collect() +} + +fn lootable(coordinator: &Coordinator, viewer_guid: u64) -> bool { + use lyracore_shared::constants::unit_dynamic_flags::LOOTABLE; + let live = coordinator.0.coord(); + let entity = live.conn.db.game_world_entity().guid().find(&1).unwrap(); + crate::codec::entity::subscriptions::projected_dynamic_flags(&live, viewer_guid, &entity) + & LOOTABLE + != 0 +} + +fn quest_state(coordinator: &Coordinator, guid: u64) -> Vec<(u32, u8)> { + coordinator + .player_quest_log(guid) + .unwrap() + .into_iter() + .map(|quest| (quest.quest_id, quest.state)) + .collect() +} + +#[test] +#[ignore = "requires the SpacetimeDB 2.7.1 CLI and Wasm toolchain"] +fn indexed_store_reads_follow_subscribe_reassignment_deletion_and_reconnect() { + let _topology = SingleShardEnv::install(); + let mut shard = Standalone::start("indexed-store"); + shard.publish_module(); + shard.assert_call("claim_operator", &[]); + shard.assert_call("install_guid_range", &["0"]); + shard.assert_call("debug_seed_scenario_fixtures", &[]); + shard.assert_call("debug_spawn_player_entity", &["1"]); + for table in [ + "game_item_instance", + "game_player_skill", + "game_character_quest", + ] { + shard.assert_sql(&format!("DELETE FROM {table}")); + } + shard.assert_call("debug_grant_item", &["1", "5090052", "2"]); + let item_guid: u64 = shard + .query_rows("SELECT guid FROM game_item_instance WHERE owner_guid = 1")[0]["guid"] + .parse() + .unwrap(); + shard.assert_call("debug_set_skill", &["1", "43", "3"]); + shard.assert_sql( + "UPDATE game_player_skill SET current = 3, max_rank = 5 WHERE character_guid = 1", + ); + shard.assert_sql("UPDATE game_quest_objective SET kind = 1, target_entry = 5090052, required_count = 2 WHERE quest_entry = 50900"); + shard.assert_call("debug_grant_quest", &["1", "50900"]); + shard.assert_sql( + "INSERT INTO game_quest_event_requirement (id,quest_entry) VALUES (9000001,50900)", + ); + shard.assert_sql("INSERT INTO game_character_quest_event_credit (id,character_guid,quest_entry) VALUES (9000001,1,50900),(9000002,1,50900)"); + shard.assert_call("debug_fill_aura_slots", &["1", "1", "false", "1"]); + shard.assert_sql("UPDATE game_aura SET eff_kind = 161, eff_p0 = 2, amount = 7, stacks = 2 WHERE target_guid = 1"); + shard.assert_sql("UPDATE game_world_entity SET dead = true, dynamic_flags = 1 WHERE guid = 1"); + shard.assert_sql("INSERT INTO game_corpse_loot_eligible (id,corpse_guid,eligible_guid) VALUES (9000001,1,1),(9000002,1,1)"); + + let runtime = tokio::runtime::Runtime::new().unwrap(); + let coordinator = connect(&shard, &runtime); + let generation = fence(&coordinator); + assert_eq!(items(&coordinator, 1), [(item_guid, 5090052, 23, 2)]); + assert!(items(&coordinator, 2).is_empty()); + assert_eq!(coordinator.player_skills(1).unwrap(), [(43, 3, 5)]); + assert_eq!(quest_state(&coordinator, 1), [(50900, 1)]); + assert_eq!( + coordinator.effective_magic_resistances(1), + [14, 0, 0, 0, 0, 0] + ); + assert!(lootable(&coordinator, 1)); + assert!(!lootable(&coordinator, 2)); + + shard.assert_sql("DELETE FROM game_character_quest_event_credit WHERE id = 9000001"); + shard.assert_sql("DELETE FROM game_corpse_loot_eligible WHERE id = 9000001"); + fence(&coordinator); + assert_eq!(quest_state(&coordinator, 1), [(50900, 1)]); + assert!(lootable(&coordinator, 1)); + + shard.assert_sql("UPDATE game_item_instance SET owner_guid = 2, slot = 24, stack_count = 3 WHERE owner_guid = 1"); + shard.assert_sql( + "UPDATE game_player_skill SET character_guid = 2, current = 4 WHERE character_guid = 1", + ); + shard.assert_sql("UPDATE game_character_quest SET character_guid = 2, quest_entry = 50901 WHERE character_guid = 1"); + shard.assert_sql( + "UPDATE game_quest_objective SET quest_entry = 50901 WHERE quest_entry = 50900", + ); + shard.assert_sql( + "UPDATE game_quest_event_requirement SET quest_entry = 50901 WHERE id = 9000001", + ); + shard.assert_sql("UPDATE game_character_quest_event_credit SET character_guid = 2, quest_entry = 50901 WHERE id = 9000002"); + shard.assert_sql( + "UPDATE game_aura SET target_guid = 2, amount = 11, stacks = 1 WHERE target_guid = 1", + ); + shard.assert_sql("UPDATE game_corpse_loot_eligible SET eligible_guid = 2 WHERE id = 9000002"); + fence(&coordinator); + assert!(items(&coordinator, 1).is_empty()); + assert_eq!(items(&coordinator, 2), [(item_guid, 5090052, 24, 3)]); + assert!(coordinator.player_skills(1).unwrap().is_empty()); + assert_eq!(coordinator.player_skills(2).unwrap(), [(43, 4, 5)]); + assert!(quest_state(&coordinator, 1).is_empty()); + assert_eq!(quest_state(&coordinator, 2), [(50901, 1)]); + assert_eq!(coordinator.effective_magic_resistances(1), [0; 6]); + assert_eq!( + coordinator.effective_magic_resistances(2), + [11, 0, 0, 0, 0, 0] + ); + assert!(!lootable(&coordinator, 1)); + assert!(lootable(&coordinator, 2)); + + shard.assert_sql("DELETE FROM game_character_quest_event_credit WHERE id = 9000002"); + fence(&coordinator); + assert_eq!(quest_state(&coordinator, 2), [(50901, 0)]); + shard.assert_sql("DELETE FROM game_quest_event_requirement WHERE id = 9000001"); + fence(&coordinator); + assert_eq!(quest_state(&coordinator, 2), [(50901, 1)]); + + for table in [ + "game_item_instance", + "game_player_skill", + "game_character_quest", + "game_quest_objective", + "game_aura", + "game_corpse_loot_eligible", + ] { + shard.assert_sql(&format!("DELETE FROM {table}")); + } + fence(&coordinator); + assert!(items(&coordinator, 2).is_empty()); + assert!(coordinator.player_skills(2).unwrap().is_empty()); + assert!(quest_state(&coordinator, 2).is_empty()); + assert_eq!(coordinator.effective_magic_resistances(2), [0; 6]); + assert!(!lootable(&coordinator, 2)); + + coordinator.park_pump_for_test(); + let old_inventory = coordinator.0.coord().inventory.clone(); + shard.assert_call("debug_grant_item", &["1", "5090052", "1"]); + shard.assert_call("debug_set_skill", &["1", "43", "1"]); + shard.assert_call("debug_grant_quest", &["1", "50900"]); + shard.assert_sql("INSERT INTO game_quest_objective (id,quest_entry,obj_index,kind,target_entry,required_count) VALUES (9000001,50900,0,1,5090052,2)"); + shard.assert_call("debug_fill_aura_slots", &["1", "1", "false", "1"]); + shard.assert_sql("UPDATE game_aura SET eff_kind = 161, eff_p0 = 2, amount = 13, stacks = 1 WHERE target_guid = 1"); + shard.assert_sql( + "INSERT INTO game_corpse_loot_eligible (id,corpse_guid,eligible_guid) VALUES (9000003,1,1)", + ); + let (reconnected, ready) = mpsc::channel(); + coordinator + .0 + .on_reconnect + .lock() + .unwrap() + .push(Arc::new(move || { + let _ = reconnected.send(()); + })); + coordinator.0.coord().conn.disconnect().unwrap(); + coordinator.resume_pump_for_test(); + ready.recv_timeout(POLL_TIMEOUT).unwrap(); + assert!(fence(&coordinator) > generation); + assert!(!Arc::ptr_eq( + &old_inventory, + &coordinator.0.coord().inventory + )); + let old_items = old_inventory.read().unwrap().item_guids(1); + let restored = items(&coordinator, 1); + assert_eq!(restored.len(), 1); + assert_eq!( + (restored[0].1, restored[0].2, restored[0].3), + (5090052, 23, 1) + ); + assert_eq!(coordinator.player_skills(1).unwrap(), [(43, 1, 5)]); + assert_eq!(quest_state(&coordinator, 1), [(50900, 0)]); + assert_eq!( + coordinator.effective_magic_resistances(1), + [13, 0, 0, 0, 0, 0] + ); + assert!(lootable(&coordinator, 1)); + shard.assert_sql("DELETE FROM game_item_instance WHERE owner_guid = 1"); + fence(&coordinator); + assert!(items(&coordinator, 1).is_empty()); + assert_eq!(old_inventory.read().unwrap().item_guids(1), old_items); +} + +#[test] +#[ignore = "requires the SpacetimeDB 2.7.1 CLI and Wasm toolchain"] +fn a_store_read_waits_until_a_real_multi_table_transaction_finishes_its_callbacks() { + let _topology = SingleShardEnv::install(); + let mut shard = Standalone::start("indexed-publication"); + shard.publish_module(); + shard.assert_call("claim_operator", &[]); + shard.assert_call("install_guid_range", &["0"]); + shard.assert_call("debug_seed_scenario_fixtures", &[]); + shard.assert_call("debug_spawn_player_entity", &["1"]); + shard.assert_sql("DELETE FROM game_item_instance WHERE owner_guid = 1"); + shard + .assert_sql("UPDATE game_quest_objective SET required_count = 1 WHERE quest_entry = 50900"); + shard.assert_call("debug_grant_quest", &["1", "50900"]); + let runtime = tokio::runtime::Runtime::new().unwrap(); + let coordinator = connect(&shard, &runtime); + assert_eq!(quest_state(&coordinator, 1), [(50900, 0)]); + let (entered, callback_entered) = mpsc::channel(); + let (release, released) = mpsc::channel(); + let released = Mutex::new(Some(released)); + coordinator + .0 + .on_pump(move |conn| { + conn.db.game_item_instance().on_insert(move |_ctx, item| { + if item.owner_guid != 1 || item.entry != 5090052 { + return; + } + if let Some(released) = released.lock().unwrap().take() { + entered.send(()).unwrap(); + released.recv_timeout(POLL_TIMEOUT).unwrap(); + } + }); + }) + .unwrap(); + shard.assert_call("debug_stress_relay", &["1", "51000", "5090052", "0"]); + callback_entered.recv_timeout(POLL_TIMEOUT).unwrap(); + let (started, reader_started) = mpsc::channel(); + let (finished, read_finished) = mpsc::channel(); + let reader = coordinator.clone(); + let read = std::thread::spawn(move || { + started.send(()).unwrap(); + finished.send(items(&reader, 1)).unwrap(); + }); + reader_started.recv_timeout(POLL_TIMEOUT).unwrap(); + let premature = read_finished.recv_timeout(Duration::from_millis(100)); + release.send(()).unwrap(); + assert!(matches!(premature, Err(mpsc::RecvTimeoutError::Timeout))); + let published = read_finished.recv_timeout(POLL_TIMEOUT).unwrap(); + read.join().unwrap(); + assert_eq!(published.len(), 1); + assert_eq!( + (published[0].1, published[0].2, published[0].3), + (5090052, 23, 1) + ); + assert!(poll_until(POLL_TIMEOUT, || quest_state(&coordinator, 1) + == [(50900, 1)])); + let log = coordinator.player_quest_log(1).unwrap(); + assert_eq!(log[0].counts, [1]); +} diff --git a/gateway/src/stdb/reads/inventory_index.rs b/gateway/src/stdb/reads/inventory_index.rs new file mode 100644 index 000000000..7d41fd99d --- /dev/null +++ b/gateway/src/stdb/reads/inventory_index.rs @@ -0,0 +1,215 @@ +use std::collections::{BTreeSet, HashMap}; +use std::sync::{Arc, RwLock}; + +use spacetimedb_sdk::{Table, TableWithPrimaryKey}; + +use super::super::bindings::*; +use super::super::connection::LiveConn; + +impl LiveConn { + pub(crate) fn item_rows(&self, owner_guid: u64) -> Vec { + self.with_cache_snapshot(|| { + self.inventory + .read() + .unwrap() + .items(&self.conn.db, owner_guid) + }) + } + + pub(crate) fn skill_rows(&self, character_guid: u64) -> Vec { + self.with_cache_snapshot(|| { + self.inventory + .read() + .unwrap() + .skills(&self.conn.db, character_guid) + }) + } +} + +#[derive(Default)] +pub(crate) struct InventoryIndex { + items: HashMap>, + skills: HashMap>, +} + +impl InventoryIndex { + fn insert_items(&mut self, row: &ItemInstance) { + self.items + .entry(row.owner_guid) + .or_default() + .insert(row.guid); + } + + fn remove_items(&mut self, row: &ItemInstance) { + if let Some(ids) = self.items.get_mut(&row.owner_guid) { + ids.remove(&row.guid); + if ids.is_empty() { + self.items.remove(&row.owner_guid); + } + } + } + + pub(crate) fn item_guids(&self, character_guid: u64) -> Vec { + self.items + .get(&character_guid) + .into_iter() + .flatten() + .copied() + .collect() + } + + pub(crate) fn items(&self, db: &RemoteTables, character_guid: u64) -> Vec { + self.item_guids(character_guid) + .into_iter() + .filter_map(|id| db.game_item_instance().guid().find(&id)) + .filter(|row| row.owner_guid == character_guid) + .collect() + } + + fn insert_skills(&mut self, row: &PlayerSkill) { + self.skills + .entry(row.character_guid) + .or_default() + .insert(row.id); + } + + fn remove_skills(&mut self, row: &PlayerSkill) { + if let Some(ids) = self.skills.get_mut(&row.character_guid) { + ids.remove(&row.id); + if ids.is_empty() { + self.skills.remove(&row.character_guid); + } + } + } + + pub(crate) fn skill_ids(&self, character_guid: u64) -> Vec { + self.skills + .get(&character_guid) + .into_iter() + .flatten() + .copied() + .collect() + } + + pub(crate) fn skills(&self, db: &RemoteTables, character_guid: u64) -> Vec { + self.skill_ids(character_guid) + .into_iter() + .filter_map(|id| db.game_player_skill().id().find(&id)) + .filter(|row| row.character_guid == character_guid) + .collect() + } +} + +pub(crate) fn watch_inventory(conn: &DbConnection) -> Arc> { + let index = Arc::new(RwLock::new(InventoryIndex::default())); + let inserted = index.clone(); + conn.db.game_item_instance().on_insert(move |_ctx, row| { + inserted.write().unwrap().insert_items(row); + }); + let updated = index.clone(); + conn.db + .game_item_instance() + .on_update(move |_ctx, old, new| { + let mut index = updated.write().unwrap(); + index.remove_items(old); + index.insert_items(new); + }); + let deleted = index.clone(); + conn.db.game_item_instance().on_delete(move |_ctx, row| { + deleted.write().unwrap().remove_items(row); + }); + let inserted = index.clone(); + conn.db.game_player_skill().on_insert(move |_ctx, row| { + inserted.write().unwrap().insert_skills(row); + }); + let updated = index.clone(); + conn.db + .game_player_skill() + .on_update(move |_ctx, old, new| { + let mut index = updated.write().unwrap(); + index.remove_skills(old); + index.insert_skills(new); + }); + let deleted = index.clone(); + conn.db.game_player_skill().on_delete(move |_ctx, row| { + deleted.write().unwrap().remove_skills(row); + }); + index +} + +#[cfg(test)] +mod tests { + use super::*; + + fn item(guid: u64, owner_guid: u64) -> ItemInstance { + ItemInstance { + guid, + owner_guid, + entry: 25, + slot: 23, + stack_count: 1, + durability: 0, + owner_identity: spacetimedb_sdk::Identity::ZERO, + created_at: spacetimedb_sdk::Timestamp::UNIX_EPOCH, + enchant_id: 0, + soulbound: false, + random_property_id: 0, + item_text_id: 0, + } + } + + fn skill(id: u64, character_guid: u64) -> PlayerSkill { + PlayerSkill { + id, + character_guid, + owner_identity: spacetimedb_sdk::Identity::ZERO, + skill_line: 43, + current: 1, + max_rank: 5, + } + } + + #[test] + fn an_item_transfer_removes_the_old_owner_and_deletion_removes_the_new_owner() { + let mut index = InventoryIndex::default(); + let old = item(7, 1); + index.insert_items(&old); + index.insert_items(&item(8, 2)); + assert_eq!(index.item_guids(1), [7]); + let moved = item(7, 2); + index.remove_items(&old); + index.insert_items(&moved); + assert!(index.item_guids(1).is_empty()); + assert_eq!(index.item_guids(2), [7, 8]); + index.remove_items(&moved); + assert_eq!(index.item_guids(2), [8]); + } + + #[test] + fn skill_reassignment_and_deletion_preserve_other_characters() { + let mut index = InventoryIndex::default(); + let old = skill(7, 1); + index.insert_skills(&old); + index.insert_skills(&skill(8, 2)); + assert_eq!(index.skill_ids(1), [7]); + let moved = skill(7, 3); + index.remove_skills(&old); + index.insert_skills(&moved); + assert!(index.skill_ids(1).is_empty()); + assert_eq!(index.skill_ids(3), [7]); + index.remove_skills(&moved); + assert!(index.skill_ids(3).is_empty()); + assert_eq!(index.skill_ids(2), [8]); + } + + #[test] + fn a_fresh_connection_rebuilds_only_its_snapshot() { + let mut old = InventoryIndex::default(); + old.insert_items(&item(7, 1)); + let mut fresh = InventoryIndex::default(); + fresh.insert_items(&item(8, 1)); + old.remove_items(&item(7, 1)); + assert_eq!(fresh.item_guids(1), [8]); + assert!(old.item_guids(1).is_empty()); + } +} diff --git a/gateway/src/stdb/reads/items.rs b/gateway/src/stdb/reads/items.rs index 07cf582cd..cdad9ddaa 100644 --- a/gateway/src/stdb/reads/items.rs +++ b/gateway/src/stdb/reads/items.rs @@ -1,5 +1,4 @@ use anyhow::Result; -use spacetimedb_sdk::Table; use super::super::bindings::*; use super::super::connection::Coordinator; @@ -8,11 +7,9 @@ use lyracore_shared::item_property::client_enchantment_id; impl Coordinator { pub fn player_skills(&self, character_guid: u64) -> Result> { let guard = self.0.coord(); - let db = &guard.conn.db; - Ok(db - .game_player_skill() - .iter() - .filter(|s| s.character_guid == character_guid) + Ok(guard + .skill_rows(character_guid) + .into_iter() .map(|s| (s.skill_line, s.current, s.max_rank)) .collect()) } @@ -26,9 +23,9 @@ impl Coordinator { fn player_item_rows(&self, owner_guid: u64) -> Vec { let guard = self.0.coord(); let db = &guard.conn.db; - db.game_item_instance() - .iter() - .filter(|i| i.owner_guid == owner_guid) + guard + .item_rows(owner_guid) + .into_iter() .filter_map(|i| { let tmpl = db.game_item_template().entry().find(&i.entry)?; Some(crate::codec::ItemInstanceView { @@ -44,15 +41,14 @@ impl Coordinator { /// Slot of `item_guid` when `owner_guid` holds it; another Character's item is `None`. pub fn item_slot_by_guid(&self, owner_guid: u64, item_guid: u64) -> Option { let guard = self.0.coord(); - let slots: Vec = guard + guard .conn .db .game_item_instance() - .iter() - .filter(|i| i.guid == item_guid && i.owner_guid == owner_guid) - .map(|i| i.slot) - .collect(); - slots.into_iter().next() + .guid() + .find(&item_guid) + .filter(|item| item.owner_guid == owner_guid) + .map(|item| item.slot) } } diff --git a/gateway/src/stdb/reads/loot_tag_index.rs b/gateway/src/stdb/reads/loot_tag_index.rs new file mode 100644 index 000000000..6710d7890 --- /dev/null +++ b/gateway/src/stdb/reads/loot_tag_index.rs @@ -0,0 +1,181 @@ +use std::collections::{BTreeSet, HashMap}; +use std::sync::{Arc, RwLock}; + +use spacetimedb_sdk::{Table, TableWithPrimaryKey}; + +use super::super::bindings::*; + +#[derive(Default)] +pub(crate) struct LootTagIndex { + tap_members: HashMap<(u64, u64), BTreeSet>, + corpse_eligible: HashMap<(u64, u64), BTreeSet>, +} + +impl LootTagIndex { + pub(crate) fn is_tap_member(&self, creature_guid: u64, character_guid: u64) -> bool { + self.tap_members + .contains_key(&(creature_guid, character_guid)) + } + + pub(crate) fn is_corpse_eligible(&self, corpse_guid: u64, character_guid: u64) -> bool { + self.corpse_eligible + .contains_key(&(corpse_guid, character_guid)) + } + + fn insert_tap_members(&mut self, row: &CreatureQuestTapMember) { + self.tap_members + .entry((row.creature_guid, row.character_guid)) + .or_default() + .insert(row.id); + } + + fn remove_tap_members(&mut self, row: &CreatureQuestTapMember) { + if let Some(ids) = self + .tap_members + .get_mut(&(row.creature_guid, row.character_guid)) + { + ids.remove(&row.id); + if ids.is_empty() { + self.tap_members + .remove(&(row.creature_guid, row.character_guid)); + } + } + } + + fn insert_corpse_eligible(&mut self, row: &CorpseLootEligible) { + self.corpse_eligible + .entry((row.corpse_guid, row.eligible_guid)) + .or_default() + .insert(row.id); + } + + fn remove_corpse_eligible(&mut self, row: &CorpseLootEligible) { + if let Some(ids) = self + .corpse_eligible + .get_mut(&(row.corpse_guid, row.eligible_guid)) + { + ids.remove(&row.id); + if ids.is_empty() { + self.corpse_eligible + .remove(&(row.corpse_guid, row.eligible_guid)); + } + } + } +} + +pub(crate) fn watch_loot_tag(conn: &DbConnection) -> Arc> { + let index = Arc::new(RwLock::new(LootTagIndex::default())); + let inserted = index.clone(); + conn.db + .game_creature_quest_tap_member() + .on_insert(move |_ctx, row| { + inserted.write().unwrap().insert_tap_members(row); + }); + let updated = index.clone(); + conn.db + .game_creature_quest_tap_member() + .on_update(move |_ctx, old, new| { + let mut index = updated.write().unwrap(); + index.remove_tap_members(old); + index.insert_tap_members(new); + }); + let deleted = index.clone(); + conn.db + .game_creature_quest_tap_member() + .on_delete(move |_ctx, row| { + deleted.write().unwrap().remove_tap_members(row); + }); + let inserted = index.clone(); + conn.db + .game_corpse_loot_eligible() + .on_insert(move |_ctx, row| { + inserted.write().unwrap().insert_corpse_eligible(row); + }); + let updated = index.clone(); + conn.db + .game_corpse_loot_eligible() + .on_update(move |_ctx, old, new| { + let mut index = updated.write().unwrap(); + index.remove_corpse_eligible(old); + index.insert_corpse_eligible(new); + }); + let deleted = index.clone(); + conn.db + .game_corpse_loot_eligible() + .on_delete(move |_ctx, row| { + deleted.write().unwrap().remove_corpse_eligible(row); + }); + index +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn tag_membership_tracks_the_creature_and_character_on_update_and_delete() { + let mut index = LootTagIndex::default(); + let old = CreatureQuestTapMember { + id: 1, + creature_guid: 99, + character_guid: 7, + }; + index.insert_tap_members(&old); + assert!(index.is_tap_member(99, 7)); + assert!(!index.is_tap_member(99, 8)); + assert!(!index.is_tap_member(100, 7)); + let moved = CreatureQuestTapMember { + id: 1, + creature_guid: 100, + character_guid: 8, + }; + index.remove_tap_members(&old); + index.insert_tap_members(&moved); + assert!(!index.is_tap_member(99, 7)); + assert!(index.is_tap_member(100, 8)); + index.remove_tap_members(&moved); + assert!(!index.is_tap_member(100, 8)); + } + + #[test] + fn corpse_eligibility_survives_a_duplicate_row_until_both_rows_leave() { + let mut index = LootTagIndex::default(); + let first = CorpseLootEligible { + id: 1, + corpse_guid: 99, + eligible_guid: 7, + }; + let second = CorpseLootEligible { + id: 2, + ..first.clone() + }; + index.insert_corpse_eligible(&first); + index.insert_corpse_eligible(&second); + index.remove_corpse_eligible(&first); + assert!(index.is_corpse_eligible(99, 7)); + assert!(!index.is_corpse_eligible(99, 8)); + assert!(!index.is_corpse_eligible(100, 7)); + index.remove_corpse_eligible(&second); + assert!(!index.is_corpse_eligible(99, 7)); + } + + #[test] + fn corpse_eligibility_moves_to_the_updated_creature_and_character() { + let mut index = LootTagIndex::default(); + let old = CorpseLootEligible { + id: 1, + corpse_guid: 99, + eligible_guid: 7, + }; + index.insert_corpse_eligible(&old); + let moved = CorpseLootEligible { + id: 1, + corpse_guid: 100, + eligible_guid: 8, + }; + index.remove_corpse_eligible(&old); + index.insert_corpse_eligible(&moved); + assert!(!index.is_corpse_eligible(99, 7)); + assert!(index.is_corpse_eligible(100, 8)); + } +} diff --git a/gateway/src/stdb/reads/mod.rs b/gateway/src/stdb/reads/mod.rs index 24294b283..6a398107a 100644 --- a/gateway/src/stdb/reads/mod.rs +++ b/gateway/src/stdb/reads/mod.rs @@ -1,36 +1,46 @@ mod account; mod channel; +mod combat_index; mod guild; +mod inventory_index; mod items; +mod loot_tag_index; mod mail; mod party; mod quest; +mod quest_index; mod spell; mod talent_reputation; mod templates; -use spacetimedb_sdk::Table; - -use super::bindings::*; - pub(crate) use quest::build_quest_log_slots; pub(crate) use account::{watch_character_names, watch_contacts, CharacterNameIndex, ContactIndex}; pub(crate) use channel::ChannelIndex; +pub(crate) use combat_index::{watch_combat, CombatIndex}; pub(crate) use guild::{watch_guilds, GuildIndex}; +pub(crate) use inventory_index::{watch_inventory, InventoryIndex}; pub(crate) use items::{property_enchant_ids, view_of_item_row}; +pub(crate) use loot_tag_index::{watch_loot_tag, LootTagIndex}; pub(crate) use mail::{watch_mail_escrows, MailEscrowIndex}; +pub(crate) use quest_index::{watch_quest, QuestIndex}; pub(in crate::stdb) use guild::petition_view; pub(in crate::stdb) use quest::quest_objectives_complete; pub(in crate::stdb) use spell::spell_ranks_stack_in_book; -pub(in crate::stdb) fn player_item_count(db: &RemoteTables, owner_guid: u64, entry: u32) -> u32 { +pub(in crate::stdb) fn player_item_count( + live: &super::connection::LiveConn, + owner_guid: u64, + entry: u32, +) -> u32 { lyracore_shared::item::carried_quantity( entry, - db.game_item_instance() - .iter() - .filter(|item| item.owner_guid == owner_guid) + live.item_rows(owner_guid) + .into_iter() .map(|item| (item.slot, item.entry, item.stack_count)), ) } + +#[cfg(test)] +mod indexed_durable_tests; diff --git a/gateway/src/stdb/reads/quest.rs b/gateway/src/stdb/reads/quest.rs index eabb90e8d..a4fa7a68f 100644 --- a/gateway/src/stdb/reads/quest.rs +++ b/gateway/src/stdb/reads/quest.rs @@ -1,26 +1,28 @@ -use spacetimedb_sdk::Table; - use super::super::bindings::*; -use super::player_item_count; +use super::super::connection::LiveConn; pub(in crate::stdb) fn quest_objectives_complete( - db: &RemoteTables, + live: &LiveConn, quest_entry: u32, logrow: Option<&CharacterQuest>, owner_guid: u64, ) -> bool { - let requires_event_credit = db - .game_quest_event_requirement() - .iter() - .any(|requirement| requirement.quest_entry == quest_entry); - let has_event_credit = db - .game_character_quest_event_credit() - .iter() - .any(|credit| credit.character_guid == owner_guid && credit.quest_entry == quest_entry); + let (objectives, requires_event_credit, has_event_credit, inventory) = live + .with_cache_snapshot(|| { + let quests = live.quests.read().unwrap(); + ( + quests.objectives(&live.conn.db, quest_entry), + quests.requires_event_credit(quest_entry), + quests.has_event_credit(owner_guid, quest_entry), + live.inventory + .read() + .unwrap() + .items(&live.conn.db, owner_guid), + ) + }); lyracore_shared::quest::objectives_complete( - db.game_quest_objective() - .iter() - .filter(|o| o.quest_entry == quest_entry) + objectives + .into_iter() .map(|o| lyracore_shared::quest::Objective { kind: o.kind, index: o.obj_index, @@ -28,20 +30,27 @@ pub(in crate::stdb) fn quest_objectives_complete( required_count: o.required_count, }), logrow.map_or(&[], |q| q.counts.as_slice()), - |entry| player_item_count(db, owner_guid, entry), + |entry| { + lyracore_shared::item::carried_quantity( + entry, + inventory + .iter() + .map(|item| (item.slot, item.entry, item.stack_count)), + ) + }, requires_event_credit, has_event_credit, ) } pub(crate) fn build_quest_log_slots( - db: &RemoteTables, - quests: &[CharacterQuest], + live: &LiveConn, player_guid: u64, ) -> Vec { - let mut active: Vec<&CharacterQuest> = quests - .iter() - .filter(|q| q.character_guid == player_guid && !q.rewarded) + let mut active: Vec = live + .quest_rows(player_guid) + .into_iter() + .filter(|q| !q.rewarded) .collect(); active.sort_by_key(|q| q.quest_entry); active @@ -49,7 +58,7 @@ pub(crate) fn build_quest_log_slots( .take(crate::codec::update_mask::idx::QUEST_LOG_SLOTS as usize) .enumerate() .map(|(i, q)| { - let complete = quest_objectives_complete(db, q.quest_entry, Some(q), player_guid); + let complete = quest_objectives_complete(live, q.quest_entry, Some(&q), player_guid); crate::codec::update_mask::QuestLogSlot { slot: i as u8, quest_id: q.quest_entry, diff --git a/gateway/src/stdb/reads/quest_index.rs b/gateway/src/stdb/reads/quest_index.rs new file mode 100644 index 000000000..c8b52ba41 --- /dev/null +++ b/gateway/src/stdb/reads/quest_index.rs @@ -0,0 +1,340 @@ +use std::collections::{BTreeSet, HashMap}; +use std::sync::{Arc, RwLock}; + +use spacetimedb_sdk::{Table, TableWithPrimaryKey}; + +use super::super::bindings::*; +use super::super::connection::LiveConn; + +impl LiveConn { + pub(crate) fn quest_rows(&self, character_guid: u64) -> Vec { + self.with_cache_snapshot(|| { + self.quests + .read() + .unwrap() + .quests(&self.conn.db, character_guid) + }) + } + + pub(crate) fn quest_objectives(&self, quest_entry: u32) -> Vec { + self.with_cache_snapshot(|| { + self.quests + .read() + .unwrap() + .objectives(&self.conn.db, quest_entry) + }) + } +} + +#[derive(Default)] +pub(crate) struct QuestIndex { + quests: HashMap>, + objectives: HashMap>, + requirements: HashMap>, + credits: HashMap<(u64, u32), BTreeSet>, +} + +impl QuestIndex { + pub(crate) fn requires_event_credit(&self, quest_entry: u32) -> bool { + self.requirements.contains_key(&quest_entry) + } + + pub(crate) fn has_event_credit(&self, character_guid: u64, quest_entry: u32) -> bool { + self.credits.contains_key(&(character_guid, quest_entry)) + } + + fn insert_quests(&mut self, row: &CharacterQuest) { + self.quests + .entry(row.character_guid) + .or_default() + .insert(row.id); + } + + fn remove_quests(&mut self, row: &CharacterQuest) { + if let Some(ids) = self.quests.get_mut(&row.character_guid) { + ids.remove(&row.id); + if ids.is_empty() { + self.quests.remove(&row.character_guid); + } + } + } + + pub(crate) fn quest_row_ids(&self, character_guid: u64) -> Vec { + self.quests + .get(&character_guid) + .into_iter() + .flatten() + .copied() + .collect() + } + + pub(crate) fn quests(&self, db: &RemoteTables, character_guid: u64) -> Vec { + self.quest_row_ids(character_guid) + .into_iter() + .filter_map(|id| db.game_character_quest().id().find(&id)) + .filter(|row| row.character_guid == character_guid) + .collect() + } + + fn insert_objectives(&mut self, row: &QuestObjective) { + self.objectives + .entry(row.quest_entry) + .or_default() + .insert(row.id); + } + + fn remove_objectives(&mut self, row: &QuestObjective) { + if let Some(ids) = self.objectives.get_mut(&row.quest_entry) { + ids.remove(&row.id); + if ids.is_empty() { + self.objectives.remove(&row.quest_entry); + } + } + } + + pub(crate) fn objective_ids(&self, quest_entry: u32) -> Vec { + self.objectives + .get(&quest_entry) + .into_iter() + .flatten() + .copied() + .collect() + } + + pub(crate) fn objectives(&self, db: &RemoteTables, quest_entry: u32) -> Vec { + self.objective_ids(quest_entry) + .into_iter() + .filter_map(|id| db.game_quest_objective().id().find(&id)) + .filter(|row| row.quest_entry == quest_entry) + .collect() + } + + fn insert_requirements(&mut self, row: &QuestEventRequirement) { + self.requirements + .entry(row.quest_entry) + .or_default() + .insert(row.id); + } + + fn remove_requirements(&mut self, row: &QuestEventRequirement) { + if let Some(ids) = self.requirements.get_mut(&row.quest_entry) { + ids.remove(&row.id); + if ids.is_empty() { + self.requirements.remove(&row.quest_entry); + } + } + } + + fn insert_credits(&mut self, row: &CharacterQuestEventCredit) { + self.credits + .entry((row.character_guid, row.quest_entry)) + .or_default() + .insert(row.id); + } + + fn remove_credits(&mut self, row: &CharacterQuestEventCredit) { + if let Some(ids) = self.credits.get_mut(&(row.character_guid, row.quest_entry)) { + ids.remove(&row.id); + if ids.is_empty() { + self.credits.remove(&(row.character_guid, row.quest_entry)); + } + } + } +} + +pub(crate) fn watch_quest(conn: &DbConnection) -> Arc> { + let index = Arc::new(RwLock::new(QuestIndex::default())); + let inserted = index.clone(); + conn.db.game_character_quest().on_insert(move |_ctx, row| { + inserted.write().unwrap().insert_quests(row); + }); + let updated = index.clone(); + conn.db + .game_character_quest() + .on_update(move |_ctx, old, new| { + let mut index = updated.write().unwrap(); + index.remove_quests(old); + index.insert_quests(new); + }); + let deleted = index.clone(); + conn.db.game_character_quest().on_delete(move |_ctx, row| { + deleted.write().unwrap().remove_quests(row); + }); + let inserted = index.clone(); + conn.db.game_quest_objective().on_insert(move |_ctx, row| { + inserted.write().unwrap().insert_objectives(row); + }); + let updated = index.clone(); + conn.db + .game_quest_objective() + .on_update(move |_ctx, old, new| { + let mut index = updated.write().unwrap(); + index.remove_objectives(old); + index.insert_objectives(new); + }); + let deleted = index.clone(); + conn.db.game_quest_objective().on_delete(move |_ctx, row| { + deleted.write().unwrap().remove_objectives(row); + }); + let inserted = index.clone(); + conn.db + .game_quest_event_requirement() + .on_insert(move |_ctx, row| { + inserted.write().unwrap().insert_requirements(row); + }); + let updated = index.clone(); + conn.db + .game_quest_event_requirement() + .on_update(move |_ctx, old, new| { + let mut index = updated.write().unwrap(); + index.remove_requirements(old); + index.insert_requirements(new); + }); + let deleted = index.clone(); + conn.db + .game_quest_event_requirement() + .on_delete(move |_ctx, row| { + deleted.write().unwrap().remove_requirements(row); + }); + let inserted = index.clone(); + conn.db + .game_character_quest_event_credit() + .on_insert(move |_ctx, row| { + inserted.write().unwrap().insert_credits(row); + }); + let updated = index.clone(); + conn.db + .game_character_quest_event_credit() + .on_update(move |_ctx, old, new| { + let mut index = updated.write().unwrap(); + index.remove_credits(old); + index.insert_credits(new); + }); + let deleted = index.clone(); + conn.db + .game_character_quest_event_credit() + .on_delete(move |_ctx, row| { + deleted.write().unwrap().remove_credits(row); + }); + index +} + +#[cfg(test)] +mod tests { + use super::*; + + fn quest(id: u64, character_guid: u64) -> CharacterQuest { + CharacterQuest { + id, + character_guid, + owner_identity: spacetimedb_sdk::Identity::ZERO, + quest_entry: 10, + counts: vec![], + rewarded: false, + deadline_micros: 0, + failed: false, + } + } + + fn objective(id: u64, quest_entry: u32) -> QuestObjective { + QuestObjective { + id, + quest_entry, + obj_index: 0, + kind: 1, + target_entry: 25, + required_count: 3, + } + } + + #[test] + fn quest_logs_and_objectives_follow_reassignment_and_deletion() { + let mut index = QuestIndex::default(); + let old = quest(7, 1); + index.insert_quests(&old); + let moved = quest(7, 2); + index.remove_quests(&old); + index.insert_quests(&moved); + assert!(index.quest_row_ids(1).is_empty()); + assert_eq!(index.quest_row_ids(2), [7]); + index.remove_quests(&moved); + assert!(index.quest_row_ids(2).is_empty()); + let old = objective(3, 10); + index.insert_objectives(&old); + index.insert_objectives(&objective(4, 10)); + let moved = objective(3, 20); + index.remove_objectives(&old); + index.insert_objectives(&moved); + assert_eq!(index.objective_ids(10), [4]); + assert_eq!(index.objective_ids(20), [3]); + index.remove_objectives(&moved); + assert!(index.objective_ids(20).is_empty()); + } + + #[test] + fn deleting_one_requirement_or_credit_preserves_another_matching_row() { + let mut index = QuestIndex::default(); + let requirement = QuestEventRequirement { + id: 1, + quest_entry: 10, + }; + let second = QuestEventRequirement { + id: 2, + quest_entry: 10, + }; + index.insert_requirements(&requirement); + index.insert_requirements(&second); + index.remove_requirements(&requirement); + assert!(index.requires_event_credit(10)); + index.remove_requirements(&second); + assert!(!index.requires_event_credit(10)); + let credit = CharacterQuestEventCredit { + id: 1, + character_guid: 7, + quest_entry: 10, + }; + let second = CharacterQuestEventCredit { + id: 2, + ..credit.clone() + }; + index.insert_credits(&credit); + index.insert_credits(&second); + index.remove_credits(&credit); + assert!(index.has_event_credit(7, 10)); + assert!(!index.has_event_credit(8, 10)); + assert!(!index.has_event_credit(7, 20)); + index.remove_credits(&second); + assert!(!index.has_event_credit(7, 10)); + } + + #[test] + fn event_requirement_and_credit_updates_drop_the_old_keys() { + let mut index = QuestIndex::default(); + let requirement = QuestEventRequirement { + id: 1, + quest_entry: 10, + }; + index.insert_requirements(&requirement); + index.remove_requirements(&requirement); + index.insert_requirements(&QuestEventRequirement { + id: 1, + quest_entry: 20, + }); + assert!(!index.requires_event_credit(10)); + assert!(index.requires_event_credit(20)); + let credit = CharacterQuestEventCredit { + id: 1, + character_guid: 7, + quest_entry: 10, + }; + index.insert_credits(&credit); + index.remove_credits(&credit); + index.insert_credits(&CharacterQuestEventCredit { + id: 1, + character_guid: 8, + quest_entry: 20, + }); + assert!(!index.has_event_credit(7, 10)); + assert!(index.has_event_credit(8, 20)); + } +} diff --git a/gateway/src/stdb/store/character.rs b/gateway/src/stdb/store/character.rs index bcedfaac4..bc1693e10 100644 --- a/gateway/src/stdb/store/character.rs +++ b/gateway/src/stdb/store/character.rs @@ -246,10 +246,10 @@ impl Coordinator { .map(character_view) .collect(); for view in &mut views { - for item in db - .game_item_instance() - .iter() - .filter(|i| i.owner_guid == view.guid && i.slot <= 18) + for item in guard + .item_rows(view.guid) + .into_iter() + .filter(|i| i.slot <= 18) { let Some(tmpl) = db.game_item_template().entry().find(&item.entry) else { continue; @@ -279,12 +279,12 @@ impl Coordinator { pub fn effective_armor(&self, guid: u64) -> u32 { let guard = self.0.coord(); - super::super::armor::effective_armor(&guard.conn.db, guid) + super::super::armor::effective_armor(&guard, guid) } pub fn effective_magic_resistances(&self, guid: u64) -> [u32; 6] { let guard = self.0.coord(); - super::super::armor::effective_magic_resistances(&guard.conn.db, guid) + super::super::armor::effective_magic_resistances(&guard, guid) } fn has_auction(&self, character_guid: u64) -> bool { diff --git a/gateway/src/stdb/store/item.rs b/gateway/src/stdb/store/item.rs index a2faa3bab..c3c0d9271 100644 --- a/gateway/src/stdb/store/item.rs +++ b/gateway/src/stdb/store/item.rs @@ -5,7 +5,6 @@ use crate::stdb::bindings::*; use crate::stdb::connection::{call_reducer, classify, DurableFailure}; use crate::stdb::Coordinator; use crate::world::{Actor, ItemActionResult, ItemActionStore, ReadableItem}; -use spacetimedb_sdk::Table; impl ItemActionStore for Coordinator { fn item_name(&self, entry: u32) -> Result> { @@ -23,10 +22,10 @@ impl ItemActionStore for Coordinator { fn readable_item(&self, owner_guid: u64, slot: u8) -> Result { let guard = self.0.coord(); let db = &guard.conn.db; - let Some(item) = db - .game_item_instance() - .iter() - .find(|i| i.owner_guid == owner_guid && i.slot == slot) + let Some(item) = guard + .item_rows(owner_guid) + .into_iter() + .find(|i| i.slot == slot) else { return Ok(ReadableItem::Missing); }; diff --git a/gateway/src/stdb/store/loot_window.rs b/gateway/src/stdb/store/loot_window.rs index daa0460a1..238e32fa0 100644 --- a/gateway/src/stdb/store/loot_window.rs +++ b/gateway/src/stdb/store/loot_window.rs @@ -147,16 +147,15 @@ fn viewer_in_raid(guard: &LiveConn, viewer_guid: u64) -> bool { fn viewer_needs_quest_item(guard: &LiveConn, viewer_guid: u64, item_entry: u32) -> bool { let db = &guard.conn.db; - let active: Vec = db - .game_character_quest() - .iter() - .filter(|quest| quest.character_guid == viewer_guid && !quest.rewarded) + let active: Vec = guard + .quest_rows(viewer_guid) + .into_iter() + .filter(|quest| !quest.rewarded) .map(|quest| quest.quest_entry) .collect(); - let objective_table = db.game_quest_objective(); - let objectives = objective_table + let objectives = active .iter() - .filter(|objective| active.contains(&objective.quest_entry)) + .flat_map(|entry| guard.quest_objectives(*entry)) .map(|objective| lyracore_shared::loot::QuestItemObjective { quest_entry: objective.quest_entry, quest_type: db @@ -173,7 +172,7 @@ fn viewer_needs_quest_item(guard: &LiveConn, viewer_guid: u64, item_entry: u32) &active, objectives, item_entry, - player_item_count(db, viewer_guid, item_entry), + player_item_count(guard, viewer_guid, item_entry), ) } diff --git a/gateway/src/stdb/store/npc.rs b/gateway/src/stdb/store/npc.rs index 8f06d6ab8..e6826bc68 100644 --- a/gateway/src/stdb/store/npc.rs +++ b/gateway/src/stdb/store/npc.rs @@ -152,9 +152,11 @@ impl NpcStore for Coordinator { let Some(template) = db.game_quest_template().entry().find(&quest_entry) else { return QuestState::Missing; }; - let Some(quest) = db.game_character_quest().iter().find(|quest| { - quest.character_guid == character_guid && quest.quest_entry == quest_entry - }) else { + let Some(quest) = guard + .quest_rows(character_guid) + .into_iter() + .find(|quest| quest.quest_entry == quest_entry) + else { return QuestState::Absent; }; if quest.rewarded { @@ -166,7 +168,7 @@ impl NpcStore for Coordinator { } else if quest.failed { QuestState::Failed } else if crate::stdb::reads::quest_objectives_complete( - db, + &guard, quest_entry, Some(&quest), character_guid, diff --git a/gateway/src/stdb/store/quest.rs b/gateway/src/stdb/store/quest.rs index 395664e44..ef142efe8 100644 --- a/gateway/src/stdb/store/quest.rs +++ b/gateway/src/stdb/store/quest.rs @@ -2,7 +2,7 @@ use anyhow::Result; use spacetimedb_sdk::Table; use crate::stdb::bindings::*; -use crate::stdb::connection::call_reducer; +use crate::stdb::connection::{call_reducer, LiveConn}; use crate::stdb::reads::{build_quest_log_slots, quest_objectives_complete}; use crate::stdb::Coordinator; use crate::world::{Actor, QuestActionStore}; @@ -11,10 +11,10 @@ impl QuestActionStore for Coordinator { fn item_start_quest(&self, owner_guid: u64, slot: u8) -> Option<(u64, u32)> { let guard = self.0.coord(); let db = &guard.conn.db; - let hit: Option<(u64, u32)> = db - .game_item_instance() - .iter() - .filter(|i| i.owner_guid == owner_guid && i.slot == slot) + let hit: Option<(u64, u32)> = guard + .item_rows(owner_guid) + .into_iter() + .filter(|i| i.slot == slot) .find_map(|i| { db.game_item_template() .entry() @@ -52,11 +52,7 @@ impl QuestActionStore for Coordinator { let player_level = player.level; let race = (player.unit_bytes_0 & 0xFF) as u8; let class = ((player.unit_bytes_0 >> 8) & 0xFF) as u8; - let log: Vec<_> = db - .game_character_quest() - .iter() - .filter(|q| q.character_guid == player_guid) - .collect(); + let log = guard.quest_rows(player_guid); let evals = match giver { Giver::Creature(entry) => db .game_creature_quest() @@ -64,7 +60,7 @@ impl QuestActionStore for Coordinator { .filter(|r| r.creature_entry == entry) .filter_map(|rel| { eval_relation( - db, + &guard, rel.quest_entry, rel.role, &log, @@ -81,7 +77,7 @@ impl QuestActionStore for Coordinator { .filter(|r| r.go_entry == entry) .filter_map(|rel| { eval_relation( - db, + &guard, rel.quest_entry, rel.role, &log, @@ -110,9 +106,7 @@ impl QuestActionStore for Coordinator { player_guid: u64, ) -> Result> { let guard = self.0.coord(); - let db = &guard.conn.db; - let quests: Vec<_> = db.game_character_quest().iter().collect(); - Ok(build_quest_log_slots(db, &quests, player_guid)) + Ok(build_quest_log_slots(&guard, player_guid)) } fn push_quest(&self, actor: Actor, quest_id: u32) -> Result<()> { @@ -224,7 +218,7 @@ pub(crate) fn quest_detail_view( #[allow(clippy::too_many_arguments)] fn eval_relation( - db: &RemoteTables, + live: &LiveConn, quest_entry: u32, role: u8, log: &[CharacterQuest], @@ -234,10 +228,11 @@ fn eval_relation( class: u8, ) -> Option { use crate::codec::{GiverQuestEval, ROLE_START}; + let db = &live.conn.db; let tmpl = db.game_quest_template().entry().find(&quest_entry)?; let logrow = log.iter().find(|q| q.quest_entry == quest_entry); let active = logrow.is_some_and(|q| !q.rewarded && !q.failed); - let complete = active && quest_objectives_complete(db, quest_entry, logrow, player_guid); + let complete = active && quest_objectives_complete(live, quest_entry, logrow, player_guid); let startable = role == ROLE_START && lyracore_shared::quest::acceptance( lyracore_shared::quest::AcceptanceRequirements { diff --git a/gateway/src/stdb/subscriptions.rs b/gateway/src/stdb/subscriptions.rs index 76dd600da..35b5df004 100644 --- a/gateway/src/stdb/subscriptions.rs +++ b/gateway/src/stdb/subscriptions.rs @@ -421,8 +421,8 @@ impl Coordinator { { let guard = self.0.coord(); let db = &guard.conn.db; - let eff = super::armor::effective_resistances(db, self_guid); - let pos = super::armor::aura_armor_positive(db, self_guid); + let eff = super::armor::effective_resistances(&guard, self_guid); + let pos = super::armor::aura_armor_positive(&guard, self_guid); let sheet = super::armor::sheet_stats(db, self_guid); drop(guard); let _ = tx.send(Outbound::One(ServerOpcodeMessage::SMSG_UPDATE_OBJECT(