2023-12-15 19:50:12 +00:00
|
|
|
package protocol
|
|
|
|
|
|
|
|
import (
|
2024-10-21 20:34:58 +00:00
|
|
|
"context"
|
2023-12-22 12:37:37 +00:00
|
|
|
"database/sql"
|
2023-12-15 19:50:12 +00:00
|
|
|
"fmt"
|
2023-12-22 12:37:37 +00:00
|
|
|
"strings"
|
2023-12-15 19:50:12 +00:00
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
|
2024-09-26 22:37:32 +00:00
|
|
|
gocommon "github.com/status-im/status-go/common"
|
2023-12-20 12:49:12 +00:00
|
|
|
"github.com/status-im/status-go/eth-node/crypto"
|
|
|
|
"github.com/status-im/status-go/protocol/common/shard"
|
2024-09-19 20:16:53 +00:00
|
|
|
"github.com/waku-org/go-waku/waku/v2/api/history"
|
2023-12-20 12:49:12 +00:00
|
|
|
|
2023-12-15 19:50:12 +00:00
|
|
|
"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
|
|
|
|
)
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
// StoreNodeRequestStats is used in tests
|
|
|
|
type StoreNodeRequestStats struct {
|
2023-12-15 19:50:12 +00:00
|
|
|
FetchedEnvelopesCount int
|
|
|
|
FetchedPagesCount int
|
|
|
|
}
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
type storeNodeRequestID struct {
|
|
|
|
RequestType storeNodeRequestType `json:"requestType"`
|
|
|
|
DataID string `json:"dataID"`
|
|
|
|
}
|
|
|
|
|
2024-03-04 20:46:25 +00:00
|
|
|
func (r *storeNodeRequestID) getCommunityID() string {
|
|
|
|
switch r.RequestType {
|
|
|
|
case storeNodeCommunityRequest:
|
|
|
|
return r.DataID
|
|
|
|
case storeNodeShardRequest:
|
|
|
|
return strings.TrimSuffix(r.DataID, transport.CommunityShardInfoTopicPrefix())
|
|
|
|
default:
|
|
|
|
return ""
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-12-15 19:50:12 +00:00
|
|
|
type StoreNodeRequestManager struct {
|
|
|
|
messenger *Messenger
|
|
|
|
logger *zap.Logger
|
|
|
|
|
|
|
|
// activeRequests contain all ongoing store node requests.
|
2023-12-20 12:49:12 +00:00
|
|
|
// Map is indexed with `DataID`.
|
|
|
|
// Request might be duplicated in the map in case of contentType collisions.
|
|
|
|
activeRequests map[storeNodeRequestID]*storeNodeRequest
|
2023-12-15 19:50:12 +00:00
|
|
|
|
|
|
|
// activeRequestsLock should be locked each time activeRequests is being accessed or changed.
|
|
|
|
activeRequestsLock sync.RWMutex
|
|
|
|
|
2024-09-19 20:16:53 +00:00
|
|
|
onPerformingBatch func(types.MailserverBatch)
|
2023-12-15 19:50:12 +00:00
|
|
|
}
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
func NewStoreNodeRequestManager(m *Messenger) *StoreNodeRequestManager {
|
2023-12-15 19:50:12 +00:00
|
|
|
return &StoreNodeRequestManager{
|
|
|
|
messenger: m,
|
|
|
|
logger: m.logger.Named("StoreNodeRequestManager"),
|
2023-12-20 12:49:12 +00:00
|
|
|
activeRequests: map[storeNodeRequestID]*storeNodeRequest{},
|
2023-12-15 19:50:12 +00:00
|
|
|
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.
|
2024-10-21 20:34:58 +00:00
|
|
|
func (m *StoreNodeRequestManager) FetchCommunity(ctx context.Context, community communities.CommunityShard, opts []StoreNodeRequestOption) (*communities.Community, StoreNodeRequestStats, error) {
|
2023-12-27 13:53:19 +00:00
|
|
|
cfg := buildStoreNodeRequestConfig(opts)
|
|
|
|
|
2023-12-15 19:50:12 +00:00
|
|
|
m.logger.Info("requesting community from store node",
|
|
|
|
zap.Any("community", community),
|
2023-12-27 13:53:19 +00:00
|
|
|
zap.Any("config", cfg))
|
2023-12-15 19:50:12 +00:00
|
|
|
|
2023-12-22 12:37:37 +00:00
|
|
|
requestCommunity := func(communityID string, shard *shard.Shard) (*communities.Community, StoreNodeRequestStats, error) {
|
2024-10-21 20:34:58 +00:00
|
|
|
channel, err := m.subscribeToRequest(ctx, storeNodeCommunityRequest, communityID, shard, cfg)
|
2023-12-22 12:37:37 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a request for community: %w", err)
|
|
|
|
}
|
|
|
|
|
2023-12-27 13:53:19 +00:00
|
|
|
if !cfg.WaitForResponse {
|
2023-12-22 12:37:37 +00:00
|
|
|
return nil, StoreNodeRequestStats{}, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
result := <-channel
|
|
|
|
return result.community, result.stats, result.err
|
2023-12-15 19:50:12 +00:00
|
|
|
}
|
|
|
|
|
2023-12-22 12:37:37 +00:00
|
|
|
// if shard was not passed or nil, request shard first
|
|
|
|
communityShard := community.Shard
|
|
|
|
if communityShard == nil {
|
|
|
|
id := transport.CommunityShardInfoTopic(community.CommunityID)
|
2024-10-21 20:34:58 +00:00
|
|
|
fetchedShard, err := m.subscribeToRequest(ctx, storeNodeShardRequest, id, shard.DefaultNonProtectedShard(), cfg)
|
2023-12-22 12:37:37 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a shard info request: %w", err)
|
|
|
|
}
|
|
|
|
|
2023-12-27 13:53:19 +00:00
|
|
|
if !cfg.WaitForResponse {
|
2023-12-22 12:37:37 +00:00
|
|
|
go func() {
|
2024-09-26 22:37:32 +00:00
|
|
|
defer gocommon.LogOnPanic()
|
2023-12-27 13:53:19 +00:00
|
|
|
shardResult := <-fetchedShard
|
2023-12-22 12:37:37 +00:00
|
|
|
communityShard = shardResult.shard
|
|
|
|
|
|
|
|
_, _, _ = requestCommunity(community.CommunityID, communityShard)
|
|
|
|
}()
|
|
|
|
return nil, StoreNodeRequestStats{}, nil
|
|
|
|
}
|
|
|
|
|
2023-12-27 13:53:19 +00:00
|
|
|
shardResult := <-fetchedShard
|
2023-12-22 12:37:37 +00:00
|
|
|
communityShard = shardResult.shard
|
2023-12-15 19:50:12 +00:00
|
|
|
}
|
|
|
|
|
2023-12-22 12:37:37 +00:00
|
|
|
// request community with on shard
|
|
|
|
return requestCommunity(community.CommunityID, communityShard)
|
2023-12-15 19:50:12 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// 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.
|
2024-10-21 20:34:58 +00:00
|
|
|
func (m *StoreNodeRequestManager) FetchCommunities(ctx context.Context, communities []communities.CommunityShard, opts []StoreNodeRequestOption) error {
|
2023-12-15 19:50:12 +00:00
|
|
|
m.logger.Info("requesting communities from store node", zap.Any("communities", communities))
|
|
|
|
|
2023-12-27 13:53:19 +00:00
|
|
|
// when fetching multiple communities we don't wait for the response
|
|
|
|
opts = append(opts, WithWaitForResponseOption(false))
|
|
|
|
|
2023-12-15 19:50:12 +00:00
|
|
|
var outErr error
|
|
|
|
|
|
|
|
for _, community := range communities {
|
2024-10-21 20:34:58 +00:00
|
|
|
_, _, err := m.FetchCommunity(ctx, community, opts)
|
2023-12-15 19:50:12 +00:00
|
|
|
if err != nil {
|
|
|
|
outErr = fmt.Errorf("%sfailed to create a request for community %s: %w", outErr, community.CommunityID, err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return outErr
|
|
|
|
}
|
|
|
|
|
2024-06-27 14:15:23 +00:00
|
|
|
// FetchContact - similar to FetchCommunity
|
|
|
|
// If a `nil` contact and a `nil` error are returned, it means that the contact wasn't found at the store node.
|
2024-10-21 20:34:58 +00:00
|
|
|
func (m *StoreNodeRequestManager) FetchContact(ctx context.Context, contactID string, opts []StoreNodeRequestOption) (*Contact, StoreNodeRequestStats, error) {
|
2023-12-27 13:53:19 +00:00
|
|
|
|
|
|
|
cfg := buildStoreNodeRequestConfig(opts)
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
m.logger.Info("requesting contact from store node",
|
|
|
|
zap.Any("contactID", contactID),
|
2023-12-27 13:53:19 +00:00
|
|
|
zap.Any("config", cfg))
|
2023-12-20 12:49:12 +00:00
|
|
|
|
2024-10-21 20:34:58 +00:00
|
|
|
channel, err := m.subscribeToRequest(ctx, storeNodeContactRequest, contactID, nil, cfg)
|
2023-12-20 12:49:12 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, StoreNodeRequestStats{}, fmt.Errorf("failed to create a request for community: %w", err)
|
|
|
|
}
|
|
|
|
|
2023-12-27 13:53:19 +00:00
|
|
|
if !cfg.WaitForResponse {
|
2023-12-20 12:49:12 +00:00
|
|
|
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
|
2023-12-15 19:50:12 +00:00
|
|
|
// a new one if not found, and returns a subscription to the result of the found/started request.
|
2023-12-20 12:49:12 +00:00
|
|
|
// The subscription can then be used to get the result of the request, this could be either a community/contact or an error.
|
2024-10-21 20:34:58 +00:00
|
|
|
func (m *StoreNodeRequestManager) subscribeToRequest(ctx context.Context, requestType storeNodeRequestType, dataID string, shard *shard.Shard, cfg StoreNodeRequestConfig) (storeNodeResponseSubscription, error) {
|
2023-12-15 19:50:12 +00:00
|
|
|
// 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()
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
requestID := storeNodeRequestID{
|
|
|
|
RequestType: requestType,
|
|
|
|
DataID: dataID,
|
|
|
|
}
|
|
|
|
|
|
|
|
request, requestFound := m.activeRequests[requestID]
|
2023-12-15 19:50:12 +00:00
|
|
|
|
|
|
|
if !requestFound {
|
|
|
|
// Create corresponding filter
|
2023-12-20 12:49:12 +00:00
|
|
|
var err error
|
|
|
|
var filter *transport.Filter
|
|
|
|
filterCreated := false
|
|
|
|
|
|
|
|
filter, filterCreated, err = m.getFilter(requestType, dataID, shard)
|
2023-12-15 19:50:12 +00:00
|
|
|
if err != nil {
|
2023-12-20 12:49:12 +00:00
|
|
|
if filterCreated {
|
|
|
|
m.forgetFilter(filter)
|
|
|
|
}
|
2023-12-15 19:50:12 +00:00
|
|
|
return nil, fmt.Errorf("failed to create community filter: %w", err)
|
|
|
|
}
|
|
|
|
|
2024-10-21 20:34:58 +00:00
|
|
|
request = m.newStoreNodeRequest(ctx)
|
2023-12-27 13:53:19 +00:00
|
|
|
request.config = cfg
|
2023-12-15 19:50:12 +00:00
|
|
|
request.pubsubTopic = filter.PubsubTopic
|
2023-12-20 12:49:12 +00:00
|
|
|
request.requestID = requestID
|
2023-12-15 19:50:12 +00:00
|
|
|
request.contentTopic = filter.ContentTopic
|
|
|
|
if filterCreated {
|
|
|
|
request.filterToForget = filter
|
|
|
|
}
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
m.activeRequests[requestID] = request
|
2023-12-15 19:50:12 +00:00
|
|
|
request.start()
|
|
|
|
}
|
|
|
|
|
|
|
|
return request.subscribe(), nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// newStoreNodeRequest creates a new storeNodeRequest struct
|
2024-10-21 20:34:58 +00:00
|
|
|
func (m *StoreNodeRequestManager) newStoreNodeRequest(ctx context.Context) *storeNodeRequest {
|
2023-12-15 19:50:12 +00:00
|
|
|
return &storeNodeRequest{
|
|
|
|
manager: m,
|
2024-10-21 20:34:58 +00:00
|
|
|
ctx: ctx,
|
2023-12-20 12:49:12 +00:00
|
|
|
subscriptions: make([]storeNodeResponseSubscription, 0),
|
2023-12-15 19:50:12 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// 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.
|
2023-12-20 12:49:12 +00:00
|
|
|
func (m *StoreNodeRequestManager) getFilter(requestType storeNodeRequestType, dataID string, shard *shard.Shard) (*transport.Filter, bool, error) {
|
2023-12-15 19:50:12 +00:00
|
|
|
// First check if such filter already exists.
|
2023-12-20 12:49:12 +00:00
|
|
|
filter := m.messenger.transport.FilterByChatID(dataID)
|
2023-12-15 19:50:12 +00:00
|
|
|
if filter != nil {
|
|
|
|
//we don't remember filter id associated with community because it was already installed
|
|
|
|
return filter, false, nil
|
|
|
|
}
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
switch requestType {
|
2024-02-20 15:49:39 +00:00
|
|
|
case storeNodeShardRequest, storeNodeCommunityRequest:
|
2023-12-20 12:49:12 +00:00
|
|
|
// 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(),
|
|
|
|
}})
|
2023-12-15 19:50:12 +00:00
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
if err != nil {
|
|
|
|
m.logger.Error("failed to install filter for community", zap.Error(err))
|
|
|
|
return nil, false, err
|
|
|
|
}
|
2023-12-15 19:50:12 +00:00
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
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)
|
2023-12-15 19:50:12 +00:00
|
|
|
}
|
|
|
|
|
2024-07-26 07:53:08 +00:00
|
|
|
filter.Ephemeral = true
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
return filter, true, nil
|
2023-12-15 19:50:12 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// 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))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
type storeNodeRequestType int
|
|
|
|
|
|
|
|
const (
|
|
|
|
storeNodeCommunityRequest storeNodeRequestType = iota
|
|
|
|
storeNodeContactRequest
|
2023-12-22 12:37:37 +00:00
|
|
|
storeNodeShardRequest
|
2023-12-20 12:49:12 +00:00
|
|
|
)
|
|
|
|
|
2023-12-15 19:50:12 +00:00
|
|
|
// 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 {
|
2023-12-20 12:49:12 +00:00
|
|
|
requestID storeNodeRequestID
|
2024-10-21 20:34:58 +00:00
|
|
|
ctx context.Context
|
2023-12-20 12:49:12 +00:00
|
|
|
|
2023-12-15 19:50:12 +00:00
|
|
|
// request parameters
|
2023-12-27 13:53:19 +00:00
|
|
|
pubsubTopic string
|
|
|
|
contentTopic types.TopicType
|
|
|
|
minimumDataClock uint64
|
|
|
|
config StoreNodeRequestConfig
|
2023-12-15 19:50:12 +00:00
|
|
|
|
|
|
|
// request corresponding metadata to be used in finalize
|
|
|
|
filterToForget *transport.Filter
|
|
|
|
|
|
|
|
// internal fields
|
|
|
|
manager *StoreNodeRequestManager
|
2023-12-20 12:49:12 +00:00
|
|
|
subscriptions []storeNodeResponseSubscription
|
|
|
|
result storeNodeRequestResult
|
2023-12-15 19:50:12 +00:00
|
|
|
}
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
// storeNodeRequestResult contains result of a single storeNodeRequest
|
|
|
|
// Further by using `data` we mean community/contact, depending on request type.
|
2023-12-15 19:50:12 +00:00
|
|
|
// If any error occurs during the request, err field will be set.
|
2023-12-20 12:49:12 +00:00
|
|
|
// 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`.
|
2023-12-15 19:50:12 +00:00
|
|
|
// stats will contain information about the performed request that might be useful for testing.
|
2023-12-20 12:49:12 +00:00
|
|
|
type storeNodeRequestResult struct {
|
|
|
|
err error
|
|
|
|
stats StoreNodeRequestStats
|
|
|
|
// One of data fields (community or contact) will be present depending on request type
|
2023-12-15 19:50:12 +00:00
|
|
|
community *communities.Community
|
2023-12-20 12:49:12 +00:00
|
|
|
contact *Contact
|
2023-12-22 12:37:37 +00:00
|
|
|
shard *shard.Shard
|
2023-12-15 19:50:12 +00:00
|
|
|
}
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
type storeNodeResponseSubscription = chan storeNodeRequestResult
|
2023-12-15 19:50:12 +00:00
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
func (r *storeNodeRequest) subscribe() storeNodeResponseSubscription {
|
|
|
|
channel := make(storeNodeResponseSubscription, 100)
|
2023-12-15 19:50:12 +00:00
|
|
|
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",
|
2023-12-20 12:49:12 +00:00
|
|
|
zap.Any("requestID", r.requestID),
|
2023-12-15 19:50:12 +00:00
|
|
|
zap.Bool("communityFound", r.result.community != nil),
|
2023-12-20 12:49:12 +00:00
|
|
|
zap.Bool("contactFound", r.result.contact != nil),
|
2023-12-22 12:37:37 +00:00
|
|
|
zap.Bool("shardFound", r.result.shard != nil),
|
2023-12-15 19:50:12 +00:00
|
|
|
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)
|
|
|
|
}
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
delete(r.manager.activeRequests, r.requestID)
|
2023-12-15 19:50:12 +00:00
|
|
|
|
|
|
|
if r.filterToForget != nil {
|
|
|
|
r.manager.forgetFilter(r.filterToForget)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2024-09-19 20:16:53 +00:00
|
|
|
func (r *storeNodeRequest) shouldFetchNextPage(envelopesCount int) (bool, uint64) {
|
2023-12-15 19:50:12 +00:00
|
|
|
logger := r.manager.logger.With(
|
2023-12-20 12:49:12 +00:00
|
|
|
zap.Any("requestID", r.requestID),
|
2023-12-15 19:50:12 +00:00
|
|
|
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
|
2023-12-20 12:49:12 +00:00
|
|
|
switch r.requestID.RequestType {
|
|
|
|
case storeNodeCommunityRequest:
|
2024-01-05 17:09:38 +00:00
|
|
|
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)
|
2023-12-15 19:50:12 +00:00
|
|
|
|
2024-01-09 21:47:37 +00:00
|
|
|
if err != nil && err != communities.ErrOrgNotFound {
|
2024-01-05 17:09:38 +00:00
|
|
|
logger.Error("failed to read community from database",
|
2023-12-20 12:49:12 +00:00
|
|
|
zap.String("communityID", r.requestID.DataID),
|
|
|
|
zap.Error(err))
|
|
|
|
r.result = storeNodeRequestResult{
|
|
|
|
community: nil,
|
2024-01-05 17:09:38 +00:00
|
|
|
err: fmt.Errorf("failed to read community from database: %w", err),
|
2023-12-20 12:49:12 +00:00
|
|
|
}
|
|
|
|
return false, 0 // failed to read from database, no sense to continue the procedure
|
2023-12-15 19:50:12 +00:00
|
|
|
}
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
if community == nil {
|
|
|
|
// community not found in the database, request next page
|
|
|
|
logger.Debug("community still not fetched")
|
2023-12-27 13:53:19 +00:00
|
|
|
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
|
2023-12-20 12:49:12 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
logger.Debug("community found",
|
|
|
|
zap.String("displayName", community.Name()))
|
2023-12-15 19:50:12 +00:00
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
r.result.community = community
|
2023-12-15 19:50:12 +00:00
|
|
|
|
2023-12-22 12:37:37 +00:00
|
|
|
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
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2024-01-23 18:20:01 +00:00
|
|
|
logger.Debug("shard found",
|
|
|
|
zap.String("community", communityIDStr),
|
|
|
|
zap.Any("shard", shardResult),
|
|
|
|
)
|
2023-12-22 12:37:37 +00:00
|
|
|
|
|
|
|
r.result.shard = shardResult
|
|
|
|
|
2023-12-20 12:49:12 +00:00
|
|
|
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")
|
2023-12-27 13:53:19 +00:00
|
|
|
return true, r.config.FurtherPageSize
|
2023-12-20 12:49:12 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
logger.Debug("contact found",
|
|
|
|
zap.String("displayName", contact.DisplayName))
|
|
|
|
|
|
|
|
r.result.contact = contact
|
|
|
|
}
|
2023-12-15 19:50:12 +00:00
|
|
|
|
2023-12-27 13:53:19 +00:00
|
|
|
return !r.config.StopWhenDataFound, r.config.FurtherPageSize
|
2023-12-15 19:50:12 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (r *storeNodeRequest) routine() {
|
2024-10-23 20:33:05 +00:00
|
|
|
defer gocommon.LogOnPanic()
|
|
|
|
|
2023-12-15 19:50:12 +00:00
|
|
|
r.manager.logger.Info("starting store node request",
|
2023-12-20 12:49:12 +00:00
|
|
|
zap.Any("requestID", r.requestID),
|
2023-12-15 19:50:12 +00:00
|
|
|
zap.String("pubsubTopic", r.pubsubTopic),
|
|
|
|
zap.Any("contentTopic", r.contentTopic),
|
|
|
|
)
|
|
|
|
|
|
|
|
// Return a nil community and no error when request was
|
2023-12-20 12:49:12 +00:00
|
|
|
// performed successfully, but no community/contact found.
|
|
|
|
r.result = storeNodeRequestResult{
|
2023-12-15 19:50:12 +00:00
|
|
|
err: nil,
|
2023-12-20 12:49:12 +00:00
|
|
|
community: nil,
|
|
|
|
contact: nil,
|
2023-12-22 12:37:37 +00:00
|
|
|
shard: nil,
|
2023-12-15 19:50:12 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
defer func() {
|
|
|
|
r.finalize()
|
|
|
|
}()
|
|
|
|
|
2024-03-04 20:46:25 +00:00
|
|
|
communityID := r.requestID.getCommunityID()
|
|
|
|
|
|
|
|
if r.requestID.RequestType != storeNodeCommunityRequest || !r.manager.messenger.communityStorenodes.HasStorenodeSetup(communityID) {
|
2024-10-21 20:34:58 +00:00
|
|
|
ctx, cancel := context.WithTimeout(r.ctx, storeNodeAvailableTimeout)
|
|
|
|
defer cancel()
|
|
|
|
if !r.manager.messenger.transport.WaitForAvailableStoreNode(ctx) {
|
2024-03-04 20:46:25 +00:00
|
|
|
r.result.err = fmt.Errorf("store node is not available")
|
|
|
|
return
|
|
|
|
}
|
2023-12-15 19:50:12 +00:00
|
|
|
}
|
2023-12-27 13:53:19 +00:00
|
|
|
|
2024-10-21 20:34:58 +00:00
|
|
|
storeNode := r.manager.messenger.getCommunityStorenode(communityID)
|
2023-12-27 13:53:19 +00:00
|
|
|
|
2024-03-04 20:46:25 +00:00
|
|
|
// Check if community already exists locally and get Clock.
|
|
|
|
if r.requestID.RequestType == storeNodeCommunityRequest {
|
|
|
|
localCommunity, _ := r.manager.messenger.communitiesManager.GetByIDString(communityID)
|
|
|
|
if localCommunity != nil {
|
|
|
|
r.minimumDataClock = localCommunity.Clock()
|
|
|
|
}
|
2023-12-27 13:53:19 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// Start store node request
|
2024-02-12 12:20:56 +00:00
|
|
|
from, to := r.manager.messenger.calculateMailserverTimeBounds(oneMonthDuration)
|
2023-12-15 19:50:12 +00:00
|
|
|
|
2024-09-19 20:16:53 +00:00
|
|
|
_, err := r.manager.messenger.performStorenodeTask(func() (*MessengerResponse, error) {
|
|
|
|
batch := types.MailserverBatch{
|
2023-12-15 19:50:12 +00:00
|
|
|
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)
|
|
|
|
}
|
|
|
|
|
2024-09-19 20:16:53 +00:00
|
|
|
return nil, r.manager.messenger.processMailserverBatchWithOptions(storeNode, batch, r.config.InitialPageSize, r.shouldFetchNextPage, true)
|
|
|
|
}, history.WithPeerID(storeNode))
|
2023-12-15 19:50:12 +00:00
|
|
|
|
|
|
|
r.result.err = err
|
|
|
|
}
|
|
|
|
|
|
|
|
func (r *storeNodeRequest) start() {
|
|
|
|
go r.routine()
|
|
|
|
}
|