<s> from twisted . internet import defer <EOL> from . appservice import ( <EOL> ApplicationServiceStore , ApplicationServiceTransactionStore <EOL> ) <EOL> from . _base import Cache <EOL> from . directory import DirectoryStore <EOL> from . events import EventsStore <EOL> from . presence import PresenceStore , UserPresenceState <EOL> from . profile import ProfileStore <EOL> from . registration import RegistrationStore <EOL> from . room import RoomStore <EOL> from . roommember import RoomMemberStore <EOL> from . stream import StreamStore <EOL> from . transactions import TransactionStore <EOL> from . keys import KeyStore <EOL> from . event_federation import EventFederationStore <EOL> from . pusher import PusherStore <EOL> from . push_rule import PushRuleStore <EOL> from . media_repository import MediaRepositoryStore <EOL> from . rejections import RejectionsStore <EOL> from . event_push_actions import EventPushActionsStore <EOL> from . state import StateStore <EOL> from . signatures import SignatureStore <EOL> from . filtering import FilteringStore <EOL> from . end_to_end_keys import EndToEndKeyStore <EOL> from . receipts import ReceiptsStore <EOL> from . search import SearchStore <EOL> from . tags import TagsStore <EOL> from . account_data import AccountDataStore <EOL> from . util . id_generators import IdGenerator , StreamIdGenerator , ChainedIdGenerator <EOL> from synapse . api . constants import PresenceState <EOL> from synapse . util . caches . stream_change_cache import StreamChangeCache <EOL> import logging <EOL> logger = logging . getLogger ( __name__ ) <EOL> LAST_SEEN_GRANULARITY = <NUM_LIT> * <NUM_LIT:1000> <EOL> class DataStore ( RoomMemberStore , RoomStore , <EOL> RegistrationStore , StreamStore , ProfileStore , <EOL> PresenceStore , TransactionStore , <EOL> DirectoryStore , KeyStore , StateStore , SignatureStore , <EOL> ApplicationServiceStore , <EOL> EventFederationStore , <EOL> MediaRepositoryStore , <EOL> RejectionsStore , <EOL> FilteringStore , <EOL> PusherStore , <EOL> PushRuleStore , <EOL> ApplicationServiceTransactionStore , <EOL> EventsStore , <EOL> ReceiptsStore , <EOL> EndToEndKeyStore , <EOL> SearchStore , <EOL> TagsStore , <EOL> AccountDataStore , <EOL> EventPushActionsStore <EOL> ) : <EOL> def __init__ ( self , db_conn , hs ) : <EOL> self . hs = hs <EOL> self . database_engine = hs . database_engine <EOL> cur = db_conn . cursor ( ) <EOL> try : <EOL> cur . execute ( "<STR_LIT>" , ) <EOL> rows = cur . fetchall ( ) <EOL> self . min_stream_token = rows [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] if rows and rows [ <NUM_LIT:0> ] and rows [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] else - <NUM_LIT:1> <EOL> self . min_stream_token = min ( self . min_stream_token , - <NUM_LIT:1> ) <EOL> finally : <EOL> cur . close ( ) <EOL> self . client_ip_last_seen = Cache ( <EOL> name = "<STR_LIT>" , <EOL> keylen = <NUM_LIT:4> , <EOL> ) <EOL> self . _stream_id_gen = StreamIdGenerator ( <EOL> db_conn , "<STR_LIT>" , "<STR_LIT>" <EOL> ) <EOL> self . _receipts_id_gen = StreamIdGenerator ( <EOL> db_conn , "<STR_LIT>" , "<STR_LIT>" <EOL> ) <EOL> self . _account_data_id_gen = StreamIdGenerator ( <EOL> db_conn , "<STR_LIT>" , "<STR_LIT>" <EOL> ) <EOL> self . _presence_id_gen = StreamIdGenerator ( <EOL> db_conn , "<STR_LIT>" , "<STR_LIT>" <EOL> ) <EOL> self . _transaction_id_gen = IdGenerator ( db_conn , "<STR_LIT>" , "<STR_LIT:id>" ) <EOL> self . _state_groups_id_gen = IdGenerator ( db_conn , "<STR_LIT>" , "<STR_LIT:id>" ) <EOL> self . _access_tokens_id_gen = IdGenerator ( db_conn , "<STR_LIT>" , "<STR_LIT:id>" ) <EOL> self . _refresh_tokens_id_gen = IdGenerator ( db_conn , "<STR_LIT>" , "<STR_LIT:id>" ) <EOL> self . _push_rule_id_gen = IdGenerator ( db_conn , "<STR_LIT>" , "<STR_LIT:id>" ) <EOL> self . _push_rules_enable_id_gen = IdGenerator ( db_conn , "<STR_LIT>" , "<STR_LIT:id>" ) <EOL> self . _push_rules_stream_id_gen = ChainedIdGenerator ( <EOL> self . _stream_id_gen , db_conn , "<STR_LIT>" , "<STR_LIT>" <EOL> ) <EOL> self . _pushers_id_gen = StreamIdGenerator ( <EOL> db_conn , "<STR_LIT>" , "<STR_LIT:id>" , <EOL> extra_tables = [ ( "<STR_LIT>" , "<STR_LIT>" ) ] , <EOL> ) <EOL> events_max = self . _stream_id_gen . get_max_token ( ) <EOL> event_cache_prefill , min_event_val = self . _get_cache_dict ( <EOL> db_conn , "<STR_LIT>" , <EOL> entity_column = "<STR_LIT>" , <EOL> stream_column = "<STR_LIT>" , <EOL> max_value = events_max , <EOL> ) <EOL> self . _events_stream_cache = StreamChangeCache ( <EOL> "<STR_LIT>" , min_event_val , <EOL> prefilled_cache = event_cache_prefill , <EOL> ) <EOL> self . _membership_stream_cache = StreamChangeCache ( <EOL> "<STR_LIT>" , events_max , <EOL> ) <EOL> account_max = self . _account_data_id_gen . get_max_token ( ) <EOL> self . _account_data_stream_cache = StreamChangeCache ( <EOL> "<STR_LIT>" , account_max , <EOL> ) <EOL> self . __presence_on_startup = self . _get_active_presence ( db_conn ) <EOL> presence_cache_prefill , min_presence_val = self . _get_cache_dict ( <EOL> db_conn , "<STR_LIT>" , <EOL> entity_column = "<STR_LIT>" , <EOL> stream_column = "<STR_LIT>" , <EOL> max_value = self . _presence_id_gen . get_max_token ( ) , <EOL> ) <EOL> self . presence_stream_cache = StreamChangeCache ( <EOL> "<STR_LIT>" , min_presence_val , <EOL> prefilled_cache = presence_cache_prefill <EOL> ) <EOL> push_rules_prefill , push_rules_id = self . _get_cache_dict ( <EOL> db_conn , "<STR_LIT>" , <EOL> entity_column = "<STR_LIT>" , <EOL> stream_column = "<STR_LIT>" , <EOL> max_value = self . _push_rules_stream_id_gen . get_max_token ( ) [ <NUM_LIT:0> ] , <EOL> ) <EOL> self . push_rules_stream_cache = StreamChangeCache ( <EOL> "<STR_LIT>" , push_rules_id , <EOL> prefilled_cache = push_rules_prefill , <EOL> ) <EOL> super ( DataStore , self ) . __init__ ( hs ) <EOL> def take_presence_startup_info ( self ) : <EOL> active_on_startup = self . __presence_on_startup <EOL> self . __presence_on_startup = None <EOL> return active_on_startup <EOL> def _get_cache_dict ( self , db_conn , table , entity_column , stream_column , max_value ) : <EOL> sql = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) % { <EOL> "<STR_LIT>" : table , <EOL> "<STR_LIT>" : entity_column , <EOL> "<STR_LIT>" : stream_column , <EOL> } <EOL> sql = self . database_engine . convert_param_style ( sql ) <EOL> txn = db_conn . cursor ( ) <EOL> txn . execute ( sql , ( int ( max_value ) , ) ) <EOL> rows = txn . fetchall ( ) <EOL> txn . close ( ) <EOL> cache = { <EOL> row [ <NUM_LIT:0> ] : int ( row [ <NUM_LIT:1> ] ) <EOL> for row in rows <EOL> } <EOL> if cache : <EOL> min_val = min ( cache . values ( ) ) <EOL> else : <EOL> min_val = max_value <EOL> return cache , min_val <EOL> def _get_active_presence ( self , db_conn ) : <EOL> """<STR_LIT>""" <EOL> sql = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> sql = self . database_engine . convert_param_style ( sql ) <EOL> txn = db_conn . cursor ( ) <EOL> txn . execute ( sql , ( PresenceState . OFFLINE , ) ) <EOL> rows = self . cursor_to_dict ( txn ) <EOL> txn . close ( ) <EOL> for row in rows : <EOL> row [ "<STR_LIT>" ] = bool ( row [ "<STR_LIT>" ] ) <EOL> return [ UserPresenceState ( ** row ) for row in rows ] <EOL> @ defer . inlineCallbacks <EOL> def insert_client_ip ( self , user , access_token , ip , user_agent ) : <EOL> now = int ( self . _clock . time_msec ( ) ) <EOL> key = ( user . to_string ( ) , access_token , ip ) <EOL> try : <EOL> last_seen = self . client_ip_last_seen . get ( key ) <EOL> except KeyError : <EOL> last_seen = None <EOL> if last_seen is not None and ( now - last_seen ) < LAST_SEEN_GRANULARITY : <EOL> defer . returnValue ( None ) <EOL> self . client_ip_last_seen . prefill ( key , now ) <EOL> yield self . _simple_upsert ( <EOL> "<STR_LIT>" , <EOL> keyvalues = { <EOL> "<STR_LIT>" : user . to_string ( ) , <EOL> "<STR_LIT>" : access_token , <EOL> "<STR_LIT>" : ip , <EOL> "<STR_LIT>" : user_agent , <EOL> } , <EOL> values = { <EOL> "<STR_LIT>" : now , <EOL> } , <EOL> desc = "<STR_LIT>" , <EOL> lock = False , <EOL> ) <EOL> @ defer . inlineCallbacks <EOL> def count_daily_users ( self ) : <EOL> """<STR_LIT>""" <EOL> def _count_users ( txn ) : <EOL> txn . execute ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> ( int ( self . _clock . time_msec ( ) ) - ( <NUM_LIT:1000> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT> ) , ) <EOL> ) <EOL> rows = self . cursor_to_dict ( txn ) <EOL> if rows : <EOL> return rows [ <NUM_LIT:0> ] [ "<STR_LIT>" ] <EOL> return <NUM_LIT:0> <EOL> ret = yield self . runInteraction ( "<STR_LIT>" , _count_users ) <EOL> defer . returnValue ( ret ) <EOL> def get_user_ip_and_agents ( self , user ) : <EOL> return self . _simple_select_list ( <EOL> table = "<STR_LIT>" , <EOL> keyvalues = { "<STR_LIT>" : user . to_string ( ) } , <EOL> retcols = [ <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" <EOL> ] , <EOL> desc = "<STR_LIT>" , <EOL> ) <EOL> def are_all_users_on_domain ( txn , database_engine , domain ) : <EOL> sql = database_engine . convert_param_style ( <EOL> "<STR_LIT>" <EOL> ) <EOL> pat = "<STR_LIT>" + domain <EOL> txn . execute ( sql , ( pat , ) ) <EOL> num_not_matching = txn . fetchall ( ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> if num_not_matching == <NUM_LIT:0> : <EOL> return True <EOL> return False </s>
