repository.py (15442B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\campaigns\client\v2\repository.py 2 from __future__ import absolute_import, division, print_function 3 from caching.client.ttl_cache import TTLCache, NO_CACHED_DATA 4 from caching.client.storage.disk_storage import BinaryDiskStorage 5 from caching.memoize import Memoize, MemoizeWithLock 6 from campaigns.client.v2.data_model.flatbuffer.event_matcher_definitions_serialization import serialize_versioned_definitions_to_flatbuffer, deserialize_flatbuffer_to_versioned_definitions_dict 7 from campaigns.client.v2.data_model.objective import ObjectiveParticipantStats, CommitmentState 8 from campaigns.client.v2.data_model.track import TrackParticipantStats 9 from campaigns.client.v2.listeners.track_listener import TrackListener 10 from campaigns.client.v2.messengers.campaigns_messenger import CampaignsMessenger 11 from campaigns.client.v2.messengers.objective_messenger import ObjectiveMessenger 12 from campaigns.client.v2.messengers.track_messenger import TrackMessenger 13 import campaigns.client.v2.signals as campaigns_signals 14 from datetime import timedelta 15 from six.moves.http_client import NOT_MODIFIED 16 from locks.singleFlight import single_flight 17 import logging 18 logger = logging.getLogger('campaigns_v2') 19 CAMPAIGNS_TTL_MINS = 15 20 CAMPAIGNS_TTL = timedelta(minutes=CAMPAIGNS_TTL_MINS) 21 STATUS_TTL_MINS = 5 22 STATUS_TTL = timedelta(minutes=STATUS_TTL_MINS) 23 24 def _is_same_version(version, other_version): 25 if version is None or other_version is None: 26 return version is other_version 27 return version == other_version 28 29 30 class CampaignsRepository(object): 31 _campaigns_messenger = None 32 _objective_messenger = None 33 _track_messenger = None 34 _track_listener = None 35 __notifyevents__ = ['OnCharacterSessionChanged'] 36 37 def __init__(self, public_gateway, service_manager, mock_public_gateway, mock_setting, campaigns_messenger_class, objective_messenger_class, track_messenger_class, track_listener_class, event_matcher_messenger_class, do_prime = True): 38 self._public_gateway = public_gateway 39 self._service_manager = service_manager 40 self._mock_public_gateway = mock_public_gateway 41 self._mock_setting = mock_setting 42 self._campaigns_messenger = campaigns_messenger_class(public_gateway) 43 self._objective_messenger = objective_messenger_class(public_gateway) 44 self._track_messenger = track_messenger_class(public_gateway) 45 self._track_listener = track_listener_class(public_gateway) 46 self._campaign_definitions_cache = TTLCache(CAMPAIGNS_TTL) 47 self._objective_definitions_cache = TTLCache(CAMPAIGNS_TTL) 48 self._campaign_status_cache = TTLCache(STATUS_TTL) 49 self._track_status_cache = TTLCache(STATUS_TTL) 50 self._objective_status_cache = TTLCache(STATUS_TTL) 51 self._objective_participant_stats_cache = TTLCache(CAMPAIGNS_TTL) 52 self._track_participant_stats_cache = TTLCache(CAMPAIGNS_TTL) 53 self._event_matcher_messenger = event_matcher_messenger_class(public_gateway) 54 self._event_matcher_cache = {} 55 self._event_matcher_disk_storage = BinaryDiskStorage('campaign_event_matcher_definitions') 56 self._update_gateway() 57 self._service_manager.RegisterNotify(self) 58 self._connect_signals() 59 if do_prime: 60 self._prime_caches_from_disk() 61 62 def __del__(self): 63 self._service_manager.UnregisterNotify(self) 64 self._disconnect_signals() 65 66 def OnCharacterSessionChanged(self, oldCharacterID, newCharacterID): 67 logger.info('OnCharacterSessionChanged') 68 self.clear_caches() 69 70 @MemoizeWithLock(CAMPAIGNS_TTL_MINS) 71 def get_all_for_namespace(self, namespace): 72 return self._campaigns_messenger.get_all_for_namespace_request(namespace) 73 74 @MemoizeWithLock(STATUS_TTL_MINS) 75 def get_all_unredeemed_participant_rewards(self, namespace): 76 unredeemed_rewards = {} 77 page_token = None 78 while True: 79 page, page_token = self._campaigns_messenger.get_all_unredeemed_participant_rewards_request(namespace, page_token) 80 for track_id, track_rewards in page: 81 if track_id in unredeemed_rewards: 82 track_rewards = unredeemed_rewards[track_id].combined_with(track_rewards) 83 unredeemed_rewards[track_id] = track_rewards 84 85 if not page_token: 86 return unredeemed_rewards 87 88 def get_track_unredeemed_rewards(self, namespace, track_id): 89 return self.get_all_unredeemed_participant_rewards(namespace).get(track_id) 90 91 def clear_caches(self): 92 self.get_all_for_namespace.clear_memoized() 93 self.get_limits.clear_memoized() 94 self._campaign_definitions_cache.clear_cache() 95 self._objective_definitions_cache.clear_cache() 96 self._campaign_status_cache.clear_cache() 97 self._objective_status_cache.clear_cache() 98 self._track_status_cache.clear_cache() 99 self._objective_participant_stats_cache.clear_cache() 100 self._track_participant_stats_cache.clear_cache() 101 self.get_all_unredeemed_participant_rewards.clear_memoized() 102 103 def _update_gateway(self, *args, **kwargs): 104 enabled = self._mock_setting.is_enabled() 105 logger.info('Mocking is enabled: %s', enabled) 106 gateway = self._mock_public_gateway if enabled else self._public_gateway 107 for messenger in (self._campaigns_messenger, 108 self._objective_messenger, 109 self._track_messenger, 110 self._event_matcher_messenger): 111 messenger.override_gateway(gateway) 112 113 self.clear_caches() 114 115 def _connect_signals(self): 116 self._mock_setting.register_to_updates(self._update_gateway) 117 self._track_listener.on_contributed.connect(self._on_track_contributed) 118 119 def _disconnect_signals(self): 120 self._mock_setting.unregister_from_updates(self._update_gateway) 121 self._track_listener.on_contributed.disconnect(self._on_track_contributed) 122 123 def _prime_caches_from_disk(self): 124 logger.info('_prime_caches_from_disk') 125 event_matcher_definitions_flatbuffer = self._event_matcher_disk_storage.get() 126 if event_matcher_definitions_flatbuffer is not None: 127 self._event_matcher_cache = deserialize_flatbuffer_to_versioned_definitions_dict(event_matcher_definitions_flatbuffer) 128 logger.info('event matcher cache: %s', self._event_matcher_cache) 129 130 def _on_track_contributed(self, track_id, total_participant_progress): 131 if track_id in self._track_status_cache: 132 self._track_status_cache.expire(track_id) 133 track_participant_stats = self._track_participant_stats_cache.get_ignore_expiry(track_id) 134 if track_participant_stats is not NO_CACHED_DATA and track_participant_stats is not None: 135 self._track_participant_stats_cache.store(track_id, TrackParticipantStats(total_participant_progress, track_participant_stats.max_participant_contribution_reached)) 136 self.get_all_unredeemed_participant_rewards.clear_memoized() 137 campaigns_signals.on_track_contributed(track_id, total_participant_progress) 138 139 @Memoize 140 def get_limits(self): 141 return self._campaigns_messenger.get_limits_request() 142 143 def commit(self, objective_id): 144 self._campaigns_messenger.commit_request(objective_id) 145 self._objective_participant_stats_cache.store(objective_id, ObjectiveParticipantStats(CommitmentState.COMMITTED)) 146 campaigns_signals.on_committed_state_changed(objective_id) 147 148 def resign(self, objective_id): 149 self._campaigns_messenger.resign_request(objective_id) 150 self._objective_participant_stats_cache.store(objective_id, ObjectiveParticipantStats(CommitmentState.RESIGNED)) 151 campaigns_signals.on_committed_state_changed(objective_id) 152 153 def redeem_participant_rewards(self, track_id): 154 self._track_messenger.redeem_participant_rewards_request(track_id) 155 self.get_all_unredeemed_participant_rewards.clear_memoized() 156 campaigns_signals.on_track_participant_rewards_redeemed(track_id) 157 158 def get_ticking_contributions_prediction(self, track_id): 159 return self._track_messenger.get_ticking_contributions_prediction_request(track_id) 160 161 @single_flight 162 def get_campaign_definition(self, campaign_id): 163 definition = self._campaign_definitions_cache.get(campaign_id) 164 if definition is not NO_CACHED_DATA: 165 return definition 166 expired_definition = self._campaign_definitions_cache.get_ignore_expiry(campaign_id) 167 known_version = None if expired_definition is NO_CACHED_DATA else expired_definition.version 168 result = self._campaigns_messenger.get_request(campaign_id, known_version) 169 if result is NOT_MODIFIED: 170 self._campaign_definitions_cache.reset_expiration(campaign_id) 171 return expired_definition 172 definition, campaign_status, track_statuses = result 173 self._campaign_definitions_cache.store(campaign_id, definition) 174 if campaign_status is not None: 175 self._campaign_status_cache.store(campaign_id, campaign_status) 176 self._store_track_statuses(track_statuses) 177 if known_version is not None: 178 self._expire_objective_definitions(definition.objective_ids) 179 campaigns_signals.on_campaign_updated(campaign_id) 180 return definition 181 182 def _store_track_statuses(self, track_statuses): 183 for track_status in track_statuses: 184 self._track_status_cache.store(track_status.track_id, track_status) 185 186 def _get_cached_track_statuses(self, track_ids): 187 track_statuses = [ self._track_status_cache.get(track_id) for track_id in track_ids ] 188 if any((track_status is NO_CACHED_DATA for track_status in track_statuses)): 189 return None 190 return track_statuses 191 192 @single_flight 193 def get_campaign_status(self, campaign_id): 194 campaign_status = self._campaign_status_cache.get(campaign_id) 195 if campaign_status is not NO_CACHED_DATA: 196 track_ids = [ track.track_id for track in self.get_campaign_definition(campaign_id).tracks ] 197 track_statuses = self._get_cached_track_statuses(track_ids) 198 if track_statuses is not None: 199 return (campaign_status, track_statuses) 200 campaign_status, track_statuses = self._campaigns_messenger.get_status_request(campaign_id) 201 self._campaign_status_cache.store(campaign_id, campaign_status) 202 self._store_track_statuses(track_statuses) 203 return (campaign_status, track_statuses) 204 205 @single_flight 206 def get_objective_status(self, objective_id, campaign_id): 207 objective_status = self._objective_status_cache.get(objective_id) 208 if objective_status is not NO_CACHED_DATA: 209 definition = self.get_objective_definition(objective_id, campaign_id) 210 track_statuses = self._get_cached_track_statuses([ track.track_id for track in definition.tracks ]) 211 if track_statuses is not None: 212 return (objective_status, track_statuses) 213 objective_status, track_statuses = self._objective_messenger.get_status_request(objective_id) 214 self._objective_status_cache.store(objective_id, objective_status) 215 self._store_track_statuses(track_statuses) 216 return (objective_status, track_statuses) 217 218 def _get_cached_track_participant_stats(self, track_ids): 219 track_participant_stats = {track_id:self._track_participant_stats_cache.get(track_id) for track_id in track_ids} 220 if any((stats is NO_CACHED_DATA for stats in track_participant_stats.values())): 221 return None 222 return track_participant_stats 223 224 @single_flight 225 def get_objective_participant_stats(self, objective_id, campaign_id): 226 objective_participant_stats = self._objective_participant_stats_cache.get(objective_id) 227 if objective_participant_stats is not NO_CACHED_DATA: 228 definition = self.get_objective_definition(objective_id, campaign_id) 229 track_participant_stats = self._get_cached_track_participant_stats([ track.track_id for track in definition.tracks ]) 230 if track_participant_stats is not None: 231 return (objective_participant_stats, track_participant_stats) 232 objective_participant_stats, track_participant_stats = self._objective_messenger.get_participant_status_request(objective_id) 233 self._objective_participant_stats_cache.store(objective_id, objective_participant_stats) 234 for track_id, stats in track_participant_stats.items(): 235 self._track_participant_stats_cache.store(track_id, stats) 236 237 return (objective_participant_stats, track_participant_stats) 238 239 @single_flight 240 def get_objective_definition(self, objective_id, campaign_id): 241 definition = self._objective_definitions_cache.get(objective_id) 242 if definition is not NO_CACHED_DATA and not self._differs_from_cached_campaign(definition, campaign_id): 243 return definition 244 expired_definition = self._objective_definitions_cache.get_ignore_expiry(objective_id) 245 known_version = None if expired_definition is NO_CACHED_DATA else expired_definition.campaign_version 246 result = self._objective_messenger.get_request(objective_id, known_version) 247 if result is NOT_MODIFIED: 248 self._objective_definitions_cache.reset_expiration(objective_id) 249 self._expire_stale_campaign_definition(campaign_id, known_version) 250 return expired_definition 251 self._objective_definitions_cache.store(objective_id, result) 252 self._expire_stale_campaign_definition(campaign_id, result.campaign_version) 253 if expired_definition is not NO_CACHED_DATA: 254 campaigns_signals.on_objective_updated(objective_id) 255 return result 256 257 def _expire_stale_campaign_definition(self, campaign_id, current_version): 258 cached_campaign = self._campaign_definitions_cache.get(campaign_id) 259 if cached_campaign is not NO_CACHED_DATA and not _is_same_version(cached_campaign.version, current_version): 260 self._campaign_definitions_cache.expire(campaign_id) 261 262 def _differs_from_cached_campaign(self, definition, campaign_id): 263 cached_campaign = self._campaign_definitions_cache.get(campaign_id) 264 if cached_campaign is NO_CACHED_DATA or cached_campaign.version is None or definition.campaign_version is None: 265 return False 266 return cached_campaign.version != definition.campaign_version 267 268 @single_flight 269 def get_event_matcher_definitions(self, semantic_version): 270 if semantic_version in self._event_matcher_cache: 271 return self._event_matcher_cache[semantic_version] 272 definitions = self._event_matcher_messenger.get_all_request(semantic_version) 273 self._event_matcher_cache[semantic_version] = definitions 274 self._update_event_matcher_disk_cache() 275 return definitions 276 277 def _update_event_matcher_disk_cache(self): 278 data = serialize_versioned_definitions_to_flatbuffer(self._event_matcher_cache) 279 self._event_matcher_disk_storage.set(data) 280 281 def _expire_objective_definitions(self, objective_ids): 282 for objective_id in objective_ids: 283 if objective_id in self._objective_definitions_cache: 284 self._objective_definitions_cache.expire(objective_id)