package protocol import ( "database/sql" "fmt" "math" "strings" "sync" "time" "github.com/status-im/status-go/eth-node/crypto" "github.com/status-im/status-go/protocol/common/shard" "go.uber.org/zap" "github.com/status-im/status-go/eth-node/types" "github.com/status-im/status-go/protocol/communities" "github.com/status-im/status-go/protocol/transport" ) const ( storeNodeAvailableTimeout = 30 * time.Second ) // StoreNodeRequestStats is used in tests type StoreNodeRequestStats struct { FetchedEnvelopesCount int FetchedPagesCount int } type storeNodeRequestID struct { RequestType storeNodeRequestType `json:"requestType"` DataID string `json:"dataID"` } type StoreNodeRequestManager struct { messenger *Messenger logger *zap.Logger // activeRequests contain all ongoing store node requests. // Map is indexed with `DataID`. // Request might be duplicated in the map in case of contentType collisions. activeRequests map[storeNodeRequestID]*storeNodeRequest // activeRequestsLock should be locked each time activeRequests is being accessed or changed. activeRequestsLock sync.RWMutex onPerformingBatch func(MailserverBatch) } func NewStoreNodeRequestManager(m *Messenger) *StoreNodeRequestManager { return &StoreNodeRequestManager{ messenger: m, logger: m.logger.Named("StoreNodeRequestManager"), activeRequests: map[storeNodeRequestID]*storeNodeRequest{}, activeRequestsLock: sync.RWMutex{}, onPerformingBatch: nil, } } // FetchCommunity makes a single request to store node for a given community id/shard pair. // When a community is successfully fetched, a `CommunityFound` event will be emitted. If `waitForResponse == true`, // the function will also wait for the store node response and return the fetched community. // Automatically waits for an available store node. // When a `nil` community and `nil` error is returned, that means the community wasn't found at the store node. func (m *StoreNodeRequestManager) FetchCommunity(community communities.CommunityShard, opts []StoreNodeRequestOption) (*communities.Community, StoreNodeRequestStats, error) { cfg := buildStoreNodeRequestConfig(opts) m.logger.Info("requesting community from store node", zap.Any("community", community), zap.Any("config", cfg)) requestCommunity := func(communityID string, shard *shard.Shard) (*communities.Community, StoreNodeRequestStats, error) { channel, err := m.subscribeToRequest(storeNodeCommunityRequest, communityID, shard, cfg) if err != nil { return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a request for community: %w", err) } if !cfg.WaitForResponse { return nil, StoreNodeRequestStats{}, nil } result := <-channel return result.community, result.stats, result.err } // if shard was not passed or nil, request shard first communityShard := community.Shard if communityShard == nil { id := transport.CommunityShardInfoTopic(community.CommunityID) fetchedShard, err := m.subscribeToRequest(storeNodeShardRequest, id, shard.DefaultShard(), cfg) if err != nil { return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a shard info request: %w", err) } if !cfg.WaitForResponse { go func() { shardResult := <-fetchedShard communityShard = shardResult.shard _, _, _ = requestCommunity(community.CommunityID, communityShard) }() return nil, StoreNodeRequestStats{}, nil } shardResult := <-fetchedShard communityShard = shardResult.shard } // request community with on shard return requestCommunity(community.CommunityID, communityShard) } // FetchCommunities makes a FetchCommunity for each element in given `communities` list. // For each successfully fetched community, a `CommunityFound` event will be emitted. Ability to subscribe // to results is not provided, because it's not needed and would complicate the code. `FetchCommunity` can // be called directly if such functionality is needed. // // This function intentionally doesn't fetch multiple content topics in a single store node request. For now // FetchCommunities is only used for regular (once in 2 minutes) fetching of curated communities. If one of // those content topics is spammed with to many envelopes, then on each iteration we will have to fetch all // of this spam first to get the envelopes in other content topics. To avoid this we keep independent requests // for each content topic. func (m *StoreNodeRequestManager) FetchCommunities(communities []communities.CommunityShard, opts []StoreNodeRequestOption) error { m.logger.Info("requesting communities from store node", zap.Any("communities", communities)) // when fetching multiple communities we don't wait for the response opts = append(opts, WithWaitForResponseOption(false)) var outErr error for _, community := range communities { _, _, err := m.FetchCommunity(community, opts) if err != nil { outErr = fmt.Errorf("%sfailed to create a request for community %s: %w", outErr, community.CommunityID, err) } } return outErr } func (m *StoreNodeRequestManager) FetchContact(contactID string, opts []StoreNodeRequestOption) (*Contact, StoreNodeRequestStats, error) { cfg := buildStoreNodeRequestConfig(opts) m.logger.Info("requesting contact from store node", zap.Any("contactID", contactID), zap.Any("config", cfg)) channel, err := m.subscribeToRequest(storeNodeContactRequest, contactID, nil, cfg) if err != nil { return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a request for community: %w", err) } if !cfg.WaitForResponse { return nil, StoreNodeRequestStats{}, nil } result := <-channel return result.contact, result.stats, result.err } // subscribeToRequest checks if a request for given community/contact is already in progress, creates and installs // a new one if not found, and returns a subscription to the result of the found/started request. // The subscription can then be used to get the result of the request, this could be either a community/contact or an error. func (m *StoreNodeRequestManager) subscribeToRequest(requestType storeNodeRequestType, dataID string, shard *shard.Shard, cfg StoreNodeRequestConfig) (storeNodeResponseSubscription, error) { // It's important to unlock only after getting the subscription channel. // We also lock `activeRequestsLock` during finalizing the requests. This ensures that the subscription // created in this function will get the result even if the requests proceeds faster than this function ends. m.activeRequestsLock.Lock() defer m.activeRequestsLock.Unlock() requestID := storeNodeRequestID{ RequestType: requestType, DataID: dataID, } request, requestFound := m.activeRequests[requestID] if !requestFound { // Create corresponding filter var err error var filter *transport.Filter filterCreated := false filter, filterCreated, err = m.getFilter(requestType, dataID, shard) if err != nil { if filterCreated { m.forgetFilter(filter) } return nil, fmt.Errorf("failed to create community filter: %w", err) } request = m.newStoreNodeRequest() request.config = cfg request.pubsubTopic = filter.PubsubTopic request.requestID = requestID request.contentTopic = filter.ContentTopic if filterCreated { request.filterToForget = filter } m.activeRequests[requestID] = request request.start() } return request.subscribe(), nil } // newStoreNodeRequest creates a new storeNodeRequest struct func (m *StoreNodeRequestManager) newStoreNodeRequest() *storeNodeRequest { return &storeNodeRequest{ manager: m, subscriptions: make([]storeNodeResponseSubscription, 0), } } // getFilter checks if a filter for a given community is already created and creates one of not found. // Returns the found/created filter, a flag if the filter was created by the function and an error. func (m *StoreNodeRequestManager) getFilter(requestType storeNodeRequestType, dataID string, shard *shard.Shard) (*transport.Filter, bool, error) { // First check if such filter already exists. filter := m.messenger.transport.FilterByChatID(dataID) if filter != nil { //we don't remember filter id associated with community because it was already installed return filter, false, nil } switch requestType { case storeNodeShardRequest: fallthrough case storeNodeCommunityRequest: // If filter wasn't installed we create it and // remember for uninstalling after response is received filters, err := m.messenger.transport.InitPublicFilters([]transport.FiltersToInitialize{{ ChatID: dataID, PubsubTopic: shard.PubsubTopic(), }}) if err != nil { m.logger.Error("failed to install filter for community", zap.Error(err)) return nil, false, err } if len(filters) != 1 { m.logger.Error("Unexpected number of filters created") return nil, true, fmt.Errorf("unexepcted number of filters created") } filter = filters[0] case storeNodeContactRequest: publicKeyBytes, err := types.DecodeHex(dataID) if err != nil { return nil, false, fmt.Errorf("failed to decode contact id: %w", err) } publicKey, err := crypto.UnmarshalPubkey(publicKeyBytes) if err != nil { return nil, false, fmt.Errorf("failed to unmarshal public key: %w", err) } filter, err = m.messenger.transport.JoinPrivate(publicKey) if err != nil { return nil, false, fmt.Errorf("failed to install filter for contact: %w", err) } default: return nil, false, fmt.Errorf("invalid store node request type: %d", requestType) } return filter, true, nil } // forgetFilter uninstalls the given filter func (m *StoreNodeRequestManager) forgetFilter(filter *transport.Filter) { err := m.messenger.transport.RemoveFilters([]*transport.Filter{filter}) if err != nil { m.logger.Warn("failed to remove filter", zap.Error(err)) } } type storeNodeRequestType int const ( storeNodeCommunityRequest storeNodeRequestType = iota storeNodeContactRequest storeNodeShardRequest ) // storeNodeRequest represents a single store node batch request. // For a valid storeNodeRequest to be performed, the user must set all the struct fields and call start method. type storeNodeRequest struct { requestID storeNodeRequestID // request parameters pubsubTopic string contentTopic types.TopicType minimumDataClock uint64 config StoreNodeRequestConfig // request corresponding metadata to be used in finalize filterToForget *transport.Filter // internal fields manager *StoreNodeRequestManager subscriptions []storeNodeResponseSubscription result storeNodeRequestResult } // storeNodeRequestResult contains result of a single storeNodeRequest // Further by using `data` we mean community/contact, depending on request type. // If any error occurs during the request, err field will be set. // If data was successfully fetched, data field will contain the fetched information. // If data wasn't found in store node, then a data will be set to `nil`. // stats will contain information about the performed request that might be useful for testing. type storeNodeRequestResult struct { err error stats StoreNodeRequestStats // One of data fields (community or contact) will be present depending on request type community *communities.Community contact *Contact shard *shard.Shard } type storeNodeResponseSubscription = chan storeNodeRequestResult func (r *storeNodeRequest) subscribe() storeNodeResponseSubscription { channel := make(storeNodeResponseSubscription, 100) r.subscriptions = append(r.subscriptions, channel) return channel } func (r *storeNodeRequest) finalize() { r.manager.activeRequestsLock.Lock() defer r.manager.activeRequestsLock.Unlock() r.manager.logger.Info("request finished", zap.Any("requestID", r.requestID), zap.Bool("communityFound", r.result.community != nil), zap.Bool("contactFound", r.result.contact != nil), zap.Bool("shardFound", r.result.shard != nil), zap.Error(r.result.err)) // Send the result to subscribers // It's important that this is done with `activeRequestsLock` locked. for _, s := range r.subscriptions { s <- r.result close(s) } if r.result.community != nil { r.manager.messenger.passStoredCommunityInfoToSignalHandler(r.result.community) } delete(r.manager.activeRequests, r.requestID) if r.filterToForget != nil { r.manager.forgetFilter(r.filterToForget) } } func (r *storeNodeRequest) shouldFetchNextPage(envelopesCount int) (bool, uint32) { logger := r.manager.logger.With( zap.Any("requestID", r.requestID), zap.Int("envelopesCount", envelopesCount)) r.result.stats.FetchedEnvelopesCount += envelopesCount r.result.stats.FetchedPagesCount++ // Force all received envelopes to be processed r.manager.messenger.ProcessAllMessages() // Try to get community from database switch r.requestID.RequestType { case storeNodeCommunityRequest: communityID, err := types.DecodeHex(r.requestID.DataID) if err != nil { logger.Error("failed to decode community ID", zap.String("communityID", r.requestID.DataID), zap.Error(err)) r.result = storeNodeRequestResult{ community: nil, err: fmt.Errorf("failed to decode community ID: %w", err), } return false, 0 // failed to decode community ID, no sense to continue the procedure } // check if community is waiting for a verification and do a verification manually _, err = r.manager.messenger.communitiesManager.ValidateCommunityByID(communityID) if err != nil { logger.Error("failed to validate community by ID", zap.String("communityID", r.requestID.DataID), zap.Error(err)) r.result = storeNodeRequestResult{ community: nil, err: fmt.Errorf("failed to validate community by ID: %w", err), } return false, 0 // failed to validate community, no sense to continue the procedure } community, err := r.manager.messenger.communitiesManager.GetByID(communityID) if err != nil && err != communities.ErrOrgNotFound { logger.Error("failed to read community from database", zap.String("communityID", r.requestID.DataID), zap.Error(err)) r.result = storeNodeRequestResult{ community: nil, err: fmt.Errorf("failed to read community from database: %w", err), } return false, 0 // failed to read from database, no sense to continue the procedure } if community == nil { // community not found in the database, request next page logger.Debug("community still not fetched") return true, r.config.FurtherPageSize } // We check here if the community was fetched actually fetched and updated, because it // could be that the community was already in the database when we started the fetching. // // Would be perfect if we could track that the community was in these particular envelopes, // but I don't think that's possible right now. We check if clock was updated instead. if community.Clock() <= r.minimumDataClock { logger.Debug("local community description is not newer than existing", zap.Any("existingClock", community.Clock()), zap.Any("minimumDataClock", r.minimumDataClock), ) return true, r.config.FurtherPageSize } logger.Debug("community found", zap.String("displayName", community.Name())) r.result.community = community case storeNodeShardRequest: communityIDStr := strings.TrimSuffix(r.requestID.DataID, transport.CommunityShardInfoTopicPrefix()) communityID, err := types.DecodeHex(communityIDStr) if err != nil { logger.Error("decode community ID failed", zap.String("communityID", communityIDStr), zap.Error(err)) return false, 0 } shardResult, err := r.manager.messenger.communitiesManager.GetCommunityShard(communityID) if err != nil { if err != sql.ErrNoRows { logger.Error("failed to read from database", zap.String("communityID", communityIDStr), zap.Error(err)) r.result = storeNodeRequestResult{ shard: nil, err: fmt.Errorf("failed to read from database: %w", err), } return false, 0 // failed to read from database, no sense to continue the procedure } } logger.Debug("shard found for ", zap.String("community", communityIDStr)) r.result.shard = shardResult case storeNodeContactRequest: contact := r.manager.messenger.GetContactByID(r.requestID.DataID) if contact == nil { // contact not found in the database, request next page logger.Debug("contact still not fetched") return true, r.config.FurtherPageSize } logger.Debug("contact found", zap.String("displayName", contact.DisplayName)) r.result.contact = contact } return !r.config.StopWhenDataFound, r.config.FurtherPageSize } func (r *storeNodeRequest) routine() { r.manager.logger.Info("starting store node request", zap.Any("requestID", r.requestID), zap.String("pubsubTopic", r.pubsubTopic), zap.Any("contentTopic", r.contentTopic), ) // Return a nil community and no error when request was // performed successfully, but no community/contact found. r.result = storeNodeRequestResult{ err: nil, community: nil, contact: nil, shard: nil, } defer func() { r.finalize() }() if !r.manager.messenger.waitForAvailableStoreNode(storeNodeAvailableTimeout) { r.result.err = fmt.Errorf("store node is not available") return } // Check if community already exists locally and get Clock. localCommunity, _ := r.manager.messenger.communitiesManager.GetByIDString(r.requestID.DataID) if localCommunity != nil { r.minimumDataClock = localCommunity.Clock() } // Start store node request to := uint32(math.Ceil(float64(r.manager.messenger.GetCurrentTimeInMillis()) / 1000)) from := to - oneMonthInSeconds _, err := r.manager.messenger.performMailserverRequest(func() (*MessengerResponse, error) { batch := MailserverBatch{ From: from, To: to, PubsubTopic: r.pubsubTopic, Topics: []types.TopicType{r.contentTopic}, } r.manager.logger.Info("perform store node request", zap.Any("batch", batch)) if r.manager.onPerformingBatch != nil { r.manager.onPerformingBatch(batch) } return nil, r.manager.messenger.processMailserverBatchWithOptions(batch, r.config.InitialPageSize, r.shouldFetchNextPage, true) }) r.result.err = err } func (r *storeNodeRequest) start() { go r.routine() }