repository.py (25055B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\campaigns\client\v1\repository.py 2 from datetime import timedelta 3 import campaigns.client.v1.data_model.flatbuffer.contribution_method_serialization as fb_contributionmethod 4 import campaigns.client.v1.signals as campaigns_signals 5 import uthread 6 import uthread2 7 from caching import Memoize 8 from caching.client.storage.disk_storage import BinaryDiskStorage 9 from caching.memoize import MemoizeWithLock 10 from caching.client.ttl_cache import TTLCache, NO_CACHED_DATA 11 from campaigns.client.v1.data_model.campaign import CampaignState 12 from campaigns.client.v1.data_model.flatbuffer.campaigndefinition_serialization import serialize_campaigns_list_to_flatbuffer, deserialize_flatbuffer_to_campaigns_list 13 from campaigns.client.v1.data_model.objective import ParticipantState, ObjectiveID 14 from campaigns.client.errors import CampaignRequestError, CampaignObjectiveGetStatsTimeoutError 15 from campaigns.client.feature_flags import campaigns_polling_frequency 16 from campaigns.client.v1.listeners.campaign_listener import CampaignListener 17 from campaigns.client.v1.listeners.objective_listener import ObjectiveListener 18 from campaigns.client.v1.messengers.campaigns_messenger import CampaignsMessenger 19 from campaigns.client.v1.messengers.contribution_methods_messenger import CampaignContributionMethodsMessenger 20 from campaigns.client.util import get_viewed_objective_ids 21 from collections import defaultdict 22 from monolithsentry.client.local_error_spike_report import record_handled_failure 23 from monolithsentry.client.error_spike_key_consts import CAMPAIGNS_OBJECTIVE_GET_STATS_TIMEOUT 24 from six.moves.http_client import NOT_MODIFIED 25 from jobboard.client import ProviderType 26 from jobboard.client.job_board_signals import on_data_primed 27 from locks import RWLock 28 from locks.singleFlight import single_flight 29 if False: 30 from typing import Dict 31 import logging 32 logger = logging.getLogger('campaigns') 33 RECONCILE_NOTICE_WAIT_SECONDS = 60 34 CAMPAIGNS_TTL_MINS = 15 35 STATS_TTL_MINS = 5 36 STATS_TTL = timedelta(minutes=STATS_TTL_MINS) 37 CAMPAIGNS_TTL = timedelta(minutes=CAMPAIGNS_TTL_MINS) 38 NO_PARTICIPATION_STATS = 'NO_PARTICIPATION_STATS' 39 40 class CampaignsRepository(object): 41 _campaigns_messenger = None 42 _contribution_methods_messenger = None 43 _campaign_listener = None 44 _objective_listener = None 45 __notifyevents__ = ['OnCharacterSessionChanged'] 46 47 def __init__(self, public_gateway, service_manager, mock_public_gateway, mock_setting, campaigns_messenger_class, contribution_methods_messenger_class, campaign_listener_class, objective_listener_class, do_data_prime = True): 48 self._public_gateway = public_gateway 49 self._service_manager = service_manager 50 self._mock_public_gateway = mock_public_gateway 51 self._mock_setting = mock_setting 52 self._poll_stats_tasklet = None 53 self._campaigns_messenger = campaigns_messenger_class(public_gateway) 54 self._contribution_methods_messenger = contribution_methods_messenger_class(public_gateway) 55 self._campaign_listener = campaign_listener_class(public_gateway) 56 self._objective_listener = objective_listener_class(public_gateway) 57 self._campaigns_cache = TTLCache(CAMPAIGNS_TTL) 58 self._campaigns_cache_disk_storage = BinaryDiskStorage('campaign_definitions') 59 self._campaign_stats_cache = TTLCache(STATS_TTL) 60 self._objective_stats_cache = TTLCache(STATS_TTL) 61 self._participant_stats_cache = TTLCache(CAMPAIGNS_TTL) 62 self._contribution_methods_cache = {} 63 self._contribution_methods_disk_storage = BinaryDiskStorage('campaign_versioned_contribution_methods') 64 if do_data_prime: 65 self._prime_caches_from_disk() 66 self._has_reconciled_caches = False 67 self._reconciliation_lock = RWLock() 68 self._prime_cache_thread = None 69 self._update_gateway() 70 self._service_manager.RegisterNotify(self) 71 self._connect_signals() 72 73 def __del__(self): 74 self._service_manager.UnregisterNotify(self) 75 self._disconnect_signals() 76 77 def OnCharacterSessionChanged(self, oldCharacterID, newCharacterID): 78 logger.info('OnCharacterSessionChanged') 79 self._participant_stats_cache.clear() 80 self.get_all_unredeemed_rewards.clear_memoized() 81 if self._has_reconciled_caches: 82 return 83 if newCharacterID is None: 84 self._clear_poll_stats_tasklet() 85 else: 86 self._start_polling_stats() 87 uthread2.start_tasklet(self._wait_to_reconcile) 88 89 def _wait_to_reconcile(self): 90 logger.info('_wait_to_reconcile') 91 uthread2.Sleep(RECONCILE_NOTICE_WAIT_SECONDS) 92 if self._has_reconciled_caches: 93 return 94 with self._reconciliation_lock: 95 if self._has_reconciled_caches: 96 return 97 self._reconcile_caches_without_version_notice() 98 99 def _reconcile_caches_without_version_notice(self): 100 logger.info('_reconcile_caches_without_version_notice') 101 current_campaign_ids = set(self.get_all_campaigns()) 102 self._reconcile_campaign_definitions(current_campaign_ids) 103 self._reconcile_contribution_methods() 104 self._has_reconciled_caches = True 105 106 def _reconcile_campaign_definitions(self, current_ids): 107 logger.info('_reconcile_campaign_definitions %s', current_ids) 108 cached_campaign_ids = set(self._campaigns_cache.keys()) 109 ids_to_remove = cached_campaign_ids - current_ids 110 for id in ids_to_remove: 111 self._campaigns_cache.remove(id) 112 113 self._update_campaign_filesystem_storage() 114 115 def _reconcile_contribution_methods(self): 116 logger.info('_reconcile_contribution_methods') 117 campaign_semantic_versions = {campaign.version.definition_semantic_verison for campaign in self._campaigns_cache.itervalues_ignore_expiry()} 118 cached_semantic_versions = set(self._contribution_methods_cache.keys()) 119 ids_to_remove = cached_semantic_versions - campaign_semantic_versions 120 for id in ids_to_remove: 121 self._contribution_methods_cache.pop(id) 122 123 self._update_contribution_methods_disk_cache() 124 125 def _prime_caches_from_disk(self): 126 logger.info('_prime_caches_from_disk') 127 campaigns_flatbuffer = self._campaigns_cache_disk_storage.get() 128 if campaigns_flatbuffer is not None: 129 campaigns_list = deserialize_flatbuffer_to_campaigns_list(campaigns_flatbuffer) 130 for campaign in campaigns_list: 131 self._campaigns_cache.store(campaign.id, campaign) 132 self._campaigns_cache.expire(campaign.id) 133 134 logger.info('campaigns cache: %s', self._campaigns_cache) 135 contribution_methods_flatbuffer = self._contribution_methods_disk_storage.get() 136 if contribution_methods_flatbuffer is not None: 137 self._contribution_methods_cache = fb_contributionmethod.deserialize_flatbuffer_to_versioned_definitions_dict(contribution_methods_flatbuffer) 138 logger.info('contribution methods cache: %s', self._contribution_methods_cache) 139 140 @MemoizeWithLock(CAMPAIGNS_TTL_MINS) 141 def get_all_campaigns(self): 142 campaign_id_list = self._campaigns_messenger.get_all_request() 143 return campaign_id_list 144 145 def get_data_for_all_current_campaigns(self): 146 campaign_id_list = self.get_all_campaigns() 147 all_data = {} 148 for campaign_id in campaign_id_list: 149 campaign_data = self.get_campaign(campaign_id) 150 if campaign_data.campaign_state == CampaignState.CANCELLED: 151 continue 152 all_data[campaign_id] = campaign_data 153 154 return all_data 155 156 @single_flight 157 def get_campaign(self, campaign_id): 158 campaign_data = self._campaigns_cache.get(campaign_id) 159 if campaign_data != NO_CACHED_DATA: 160 return campaign_data 161 version = None 162 expired_campaign_data = self._campaigns_cache.get_ignore_expiry(campaign_id) 163 if expired_campaign_data != NO_CACHED_DATA: 164 version = expired_campaign_data.version 165 result = self._campaigns_messenger.get_request(campaign_id, version) 166 if result == NOT_MODIFIED: 167 self._campaigns_cache.reset_expiration(campaign_id) 168 return expired_campaign_data 169 self._campaigns_cache.store(result.id, result) 170 self._update_campaign_filesystem_storage() 171 if version is not None: 172 uthread2.start_tasklet(self._emit_on_campaign_updated, campaign_id) 173 return result 174 175 def _update_campaign_filesystem_storage(self): 176 campaigns = [ campaign for campaign in self._campaigns_cache.itervalues_ignore_expiry() if campaign != NO_CACHED_DATA ] 177 data = serialize_campaigns_list_to_flatbuffer(campaigns) 178 self._campaigns_cache_disk_storage.set(data) 179 180 @single_flight 181 def get_campaign_stats(self, campaign_id): 182 campaign_stats = self._campaign_stats_cache.get(campaign_id) 183 objective_out_of_date = False 184 objective_stats = [] 185 if campaign_stats != NO_CACHED_DATA: 186 campaign = self.get_campaign(campaign_id) 187 current_objectives = campaign.current_objectives 188 for objective in current_objectives: 189 objective_stat = self._objective_stats_cache.get(objective.id) 190 if objective_stat == NO_CACHED_DATA: 191 objective_out_of_date = True 192 break 193 else: 194 objective_stats.append(objective_stat) 195 196 if campaign_stats != NO_CACHED_DATA and not objective_out_of_date: 197 return (campaign_stats, objective_stats) 198 campaign_stats, objective_stats = self._campaigns_messenger.get_stats_request(campaign_id) 199 self._campaign_stats_cache.store(campaign_stats.id, campaign_stats) 200 for objective_stat in objective_stats: 201 self._objective_stats_cache.store(objective_stat.objective_id, objective_stat) 202 203 return (campaign_stats, objective_stats) 204 205 @single_flight 206 def get_campaign_participant_stats(self, campaign_id): 207 new_data_required = False 208 participant_stats = [] 209 campaign = self.get_campaign(campaign_id) 210 current_objectives = campaign.current_objectives 211 for objective in current_objectives: 212 participant_stat = self._participant_stats_cache.get(objective.id) 213 if participant_stat == NO_CACHED_DATA: 214 new_data_required = True 215 elif participant_stat != NO_PARTICIPATION_STATS: 216 participant_stats.append(participant_stat) 217 218 if not new_data_required: 219 return participant_stats 220 results = self._campaigns_messenger.get_participant_stats_request(campaign_id) 221 stat_by_objective_id = {participant_stat.objective_id:participant_stat for participant_stat in results} 222 for objective in current_objectives: 223 if objective.id in stat_by_objective_id: 224 self._participant_stats_cache.store(objective.id, stat_by_objective_id[objective.id]) 225 else: 226 self._participant_stats_cache.store(objective.id, NO_PARTICIPATION_STATS) 227 228 return results 229 230 @MemoizeWithLock(STATS_TTL_MINS) 231 def get_all_unredeemed_rewards(self): 232 next_page_token = None 233 rewards_by_objective_id = defaultdict(list) 234 while True: 235 result, next_page_token = self._campaigns_messenger.get_all_unredeemed_rewards_request(next_page_token) 236 for r in result: 237 rewards_by_objective_id[r.objective_id].append(r) 238 239 if not next_page_token: 240 break 241 242 return rewards_by_objective_id 243 244 def get_unredeemed_rewards_for_objective(self, objective_id): 245 return self.get_all_unredeemed_rewards().get(objective_id) 246 247 @Memoize 248 def get_limits(self): 249 return self._campaigns_messenger.get_limits_request() 250 251 @single_flight 252 def get_objective_stats(self, objective_id_list): 253 expired_ids = [] 254 cached_objective_stats = [] 255 for objective_id in objective_id_list: 256 objective_stat = self._objective_stats_cache.get(objective_id) 257 if objective_stat == NO_CACHED_DATA: 258 expired_ids.append(objective_id) 259 else: 260 cached_objective_stats.append(objective_stat) 261 262 if len(expired_ids) == 0: 263 return cached_objective_stats 264 results_list = self._campaigns_messenger.get_objective_stats_request(expired_ids) 265 updated_objective_stats = set() 266 for objective_stat in results_list: 267 self._objective_stats_cache.store(objective_stat.objective_id, objective_stat) 268 updated_objective_stats.add(objective_stat.objective_id) 269 270 if updated_objective_stats: 271 uthread2.start_tasklet(self._emit_on_campaigns_objective_stats_updated, updated_objective_stats) 272 cached_objective_stats.extend(results_list) 273 return cached_objective_stats 274 275 def objective_stats_have_valid_cache(self, objective_id): 276 return self._objective_stats_cache.has_cached_data_for_key(objective_id) 277 278 def objective_participant_stats_have_valid_cache(self, objective_id): 279 return self._participant_stats_cache.has_cached_data_for_key(objective_id) 280 281 def _emit_on_campaigns_objective_stats_updated(self, updated_objective_stats): 282 campaigns_signals.on_campaigns_objective_stats_updated(updated_objective_stats) 283 284 def _emit_on_campaign_updated(self, campaign_id): 285 logger.info('_emit_on_campaign_updated %s', campaign_id) 286 campaigns_signals.on_campaign_updated(campaign_id) 287 288 @single_flight 289 def get_contribution_method_definitions(self, semantic_version): 290 if semantic_version in self._contribution_methods_cache: 291 return self._contribution_methods_cache[semantic_version] 292 definitions = self._contribution_methods_messenger.get_all_request(semantic_version) 293 self._contribution_methods_cache[semantic_version] = definitions 294 self._update_contribution_methods_disk_cache() 295 return definitions 296 297 def _update_contribution_methods_disk_cache(self): 298 data = fb_contributionmethod.serialize_versioned_definitions_to_flatbuffer(self._contribution_methods_cache) 299 self._contribution_methods_disk_storage.set(data) 300 301 def commit(self, objective_id): 302 self._campaigns_messenger.commit_request(objective_id) 303 if objective_id not in self._participant_stats_cache: 304 logger.warning("Committed to an objective for which we don't have any participant stats") 305 return 306 participant_stat = self._participant_stats_cache.get_ignore_expiry(objective_id) 307 if participant_stat == NO_PARTICIPATION_STATS: 308 self._participant_stats_cache.remove(objective_id) 309 else: 310 participant_stat.participant_state = ParticipantState.COMMITTED 311 self._participant_stats_cache.reset_expiration(objective_id) 312 campaigns_signals.on_commited_state_changed(objective_id) 313 314 def resign(self, objective_id): 315 self._campaigns_messenger.resign_request(objective_id) 316 if objective_id not in self._participant_stats_cache: 317 logger.warning("Resigning from an objective for which we don't have any participant stats") 318 return 319 participant_stat = self._participant_stats_cache.get_ignore_expiry(objective_id) 320 if participant_stat == NO_PARTICIPATION_STATS: 321 self._participant_stats_cache.remove(objective_id) 322 else: 323 participant_stat.participant_state = ParticipantState.RESIGNED 324 self._participant_stats_cache.reset_expiration(objective_id) 325 campaigns_signals.on_commited_state_changed(objective_id) 326 327 def redeem_rewards(self, objective_id): 328 try: 329 self._campaigns_messenger.redeem_rewards_request(objective_id) 330 self.get_all_unredeemed_rewards.clear_memoized() 331 except CampaignRequestError as e: 332 logger.exception('Failed to redeem rewards for objective') 333 campaigns_signals.on_redeem_reward_failed(objective_id) 334 else: 335 campaigns_signals.on_reward_redeemed(objective_id) 336 337 def clear_caches(self): 338 self.get_all_campaigns.clear_memoized() 339 self.get_limits.clear_memoized() 340 self.get_all_unredeemed_rewards.clear_memoized() 341 self._campaigns_cache.clear_cache() 342 self._campaigns_cache_disk_storage.clear() 343 self._campaign_stats_cache.clear_cache() 344 self._objective_stats_cache.clear_cache() 345 self._participant_stats_cache.clear_cache() 346 self._contribution_methods_cache.clear() 347 self._contribution_methods_disk_storage.clear() 348 campaigns_signals.on_campaigns_cache_invalidated() 349 350 def clear_stats_caches(self): 351 self._campaign_stats_cache.clear_cache() 352 self._objective_stats_cache.clear_cache() 353 self._participant_stats_cache.clear_cache() 354 355 def clear_campaign_stats_cache(self, campaign_id): 356 self._campaign_stats_cache.pop(campaign_id, None) 357 358 def _update_gateway(self, *args, **kwargs): 359 enabled = self._mock_setting.is_enabled() 360 logger.info('Mocking is enabled: %s', enabled) 361 if enabled: 362 self._campaigns_messenger.override_gateway(self._mock_public_gateway) 363 self._contribution_methods_messenger.override_gateway(self._mock_public_gateway) 364 else: 365 self._campaigns_messenger.override_gateway(self._public_gateway) 366 self._contribution_methods_messenger.override_gateway(self._public_gateway) 367 368 def _on_campaign_started(self, campaign_id): 369 campaigns_signals.on_campaign_started(campaign_id) 370 self.get_all_campaigns.clear_memoized() 371 372 def _on_campaign_completed(self, campaign_id): 373 campaigns_signals.on_campaign_completed(campaign_id) 374 self.get_all_campaigns.clear_memoized() 375 376 def _on_campaign_expired(self, campaign_id): 377 campaigns_signals.on_campaign_expired(campaign_id) 378 self.get_all_campaigns.clear_memoized() 379 380 def _on_campaigns_version(self, versions_dict): 381 campaigns_signals.on_campaigns_version(versions_dict) 382 with self._reconciliation_lock: 383 current_campaign_ids = set(versions_dict.keys()) 384 self._reconcile_campaign_definitions(current_campaign_ids) 385 self._reconcile_contribution_methods() 386 self._reconcile_campaign_versions(versions_dict) 387 self._has_reconciled_caches = True 388 389 def _reconcile_campaign_versions(self, versions_dict): 390 logger.info('_reconcile_campaign_versions %s', versions_dict) 391 for campaign_id, campaign in self._campaigns_cache.iteritems(): 392 if campaign_id not in versions_dict: 393 logger.warning('Campaign ID not in version notice, this should have been reconciled already %s', campaign_id) 394 continue 395 version = versions_dict[campaign_id] 396 if version != campaign.version: 397 self._campaigns_cache.expire(campaign_id) 398 399 def _on_objective_contributed_notice(self, objective_id, total_participant_progress): 400 if objective_id not in self._participant_stats_cache: 401 logger.warning('Received contribution notice with no participant stats for objective id %s', objective_id) 402 return 403 participant_stats = self._participant_stats_cache.get_ignore_expiry(objective_id) 404 if participant_stats == NO_PARTICIPATION_STATS: 405 logger.warning('Got unexpected NO_PARTICIPATION_STATS in cache for objective contribution %s', objective_id) 406 self._participant_stats_cache.remove(objective_id) 407 else: 408 participant_stats.current_participant_progress = total_participant_progress 409 self._participant_stats_cache.reset_expiration(objective_id) 410 if objective_id in self._objective_stats_cache: 411 self._objective_stats_cache.expire(objective_id) 412 self.get_all_unredeemed_rewards.clear_memoized() 413 campaigns_signals.on_objective_contributed_notice(objective_id, total_participant_progress) 414 415 def _connect_signals(self): 416 self._mock_setting.register_to_updates(self._update_gateway) 417 self._campaign_listener.on_started.connect(self._on_campaign_started) 418 self._campaign_listener.on_completed.connect(self._on_campaign_completed) 419 self._campaign_listener.on_expired.connect(self._on_campaign_expired) 420 self._campaign_listener.on_current_versions.connect(self._on_campaigns_version) 421 self._objective_listener.on_contributed.connect(self._on_objective_contributed_notice) 422 campaigns_signals.on_campaigns_polling_frequency_changed.connect(self._on_polling_frequency_changed) 423 424 def _disconnect_signals(self): 425 self._mock_setting.unregister_from_updates(self._update_gateway) 426 self._campaign_listener.on_started.disconnect(self._on_campaign_started) 427 self._campaign_listener.on_completed.disconnect(self._on_campaign_completed) 428 self._campaign_listener.on_expired.disconnect(self._on_campaign_expired) 429 self._campaign_listener.on_current_versions.disconnect(self._on_campaigns_version) 430 self._objective_listener.on_contributed.disconnect(self._on_objective_contributed_notice) 431 campaigns_signals.on_campaigns_polling_frequency_changed.disconnect(self._on_polling_frequency_changed) 432 433 def get_campaign_objectives(self, campaign_id): 434 campaign_data = self.get_campaign(campaign_id) 435 if campaign_data.campaign_state == CampaignState.CANCELLED: 436 return [] 437 return campaign_data.current_objectives + campaign_data.future_objectives 438 439 def get_all_campaigns_objectives(self): 440 all_current_objectives = [] 441 all_future_objectives = [] 442 all_campaign_ids = self.get_all_campaigns() 443 for campaign_id in all_campaign_ids: 444 campaign_data = self.get_campaign(campaign_id) 445 if campaign_data.campaign_state == CampaignState.CANCELLED: 446 continue 447 current_objectives = campaign_data.current_objectives 448 for obj in current_objectives: 449 all_current_objectives.append(obj) 450 451 future_objectives = campaign_data.future_objectives 452 for obj in future_objectives: 453 all_future_objectives.append(obj) 454 455 return all_current_objectives + all_future_objectives 456 457 def _on_polling_frequency_changed(self, *args, **kwargs): 458 self._clear_poll_stats_tasklet() 459 self._start_polling_stats() 460 461 def _get_polling_interval(self): 462 return campaigns_polling_frequency() 463 464 def get_viewed_ids(self): 465 viewed_objective_ids = get_viewed_objective_ids() 466 return viewed_objective_ids 467 468 def _start_polling_stats(self): 469 if self._poll_stats_tasklet is None: 470 self._poll_stats_tasklet = uthread2.start_tasklet(self._poll_stats) 471 472 def _poll_stats(self): 473 while True: 474 frequency_sec = self._get_polling_interval() 475 uthread2.sleep(frequency_sec) 476 viewed_ids = self.get_viewed_ids() 477 self._request_batched_stats(viewed_ids) 478 479 def _request_batched_stats(self, project_ids): 480 try: 481 self.get_objective_stats(project_ids) 482 except CampaignObjectiveGetStatsTimeoutError as exc: 483 record_handled_failure(CAMPAIGNS_OBJECTIVE_GET_STATS_TIMEOUT, 'Objective stats poll timed out: %s' % exc) 484 except Exception: 485 logger.exception('Failed when getting objective stats') 486 487 def _clear_poll_stats_tasklet(self): 488 if self._poll_stats_tasklet: 489 self._poll_stats_tasklet.kill() 490 self._poll_stats_tasklet = None 491 492 def on_frequency_changed(self, *args, **kwargs): 493 self._clear_poll_stats_tasklet() 494 self._start_polling_stats() 495 496 def prime_stats_for_campaign(self, campaign_id): 497 self.get_campaign_stats(campaign_id) 498 self.get_campaign_participant_stats(campaign_id) 499 500 def prime_all_stats_for_campaign(self, campaign_id): 501 if campaign_id not in self.get_data_for_all_current_campaigns(): 502 return 503 self.prime_all_stats() 504 505 def prime_all_stats(self): 506 if self._prime_cache_thread and self._prime_cache_thread.is_alive(): 507 return 508 self._prime_cache_thread = uthread2.StartTasklet(self._prime_all_stats) 509 510 def _prime_all_stats(self): 511 all_current_campaigns_by_id = self.get_data_for_all_current_campaigns() 512 calls = [ (self.get_campaign_stats, (campaign_id,)) for campaign_id in all_current_campaigns_by_id.iterkeys() ] 513 calls += [ (self.get_campaign_participant_stats, (campaign_id,)) for campaign_id in all_current_campaigns_by_id.iterkeys() ] 514 if calls: 515 uthread.parallel(calls) 516 on_data_primed(ProviderType.CAMPAIGNS)