firebase-ios-sdk/Firebase/Database/Core/FPersistentConnection.m

/*
 * Copyright 2017 Google
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#import <Foundation/Foundation.h>

#import <FirebaseCore/FIRLogger.h>
#import <SystemConfiguration/SystemConfiguration.h>
#import <netinet/in.h>
#import <dlfcn.h>
#import "FIRDatabaseReference.h"
#import "FPersistentConnection.h"
#import "FConstants.h"
#import "FAtomicNumber.h"
#import "FQueryParams.h"
#import "FTupleOnDisconnect.h"
#import "FTupleCallbackStatus.h"
#import "FQuerySpec.h"
#import "FIndex.h"
#import "FIRDatabaseConfig.h"
#import "FIRDatabaseConfig_Private.h"
#import "FSnapshotUtilities.h"
#import "FRangeMerge.h"
#import "FCompoundHash.h"
#import "FSyncTree.h"
#import "FIRRetryHelper.h"
#import "FAuthTokenProvider.h"
#import "FUtilities.h"

@interface FOutstandingQuery : NSObject

@property (nonatomic, strong) FQuerySpec* query;
@property (nonatomic, strong) NSNumber *tagId;
@property (nonatomic, strong) id<FSyncTreeHash> syncTreeHash;
@property (nonatomic, copy) fbt_void_nsstring onComplete;

@end

@implementation FOutstandingQuery

@end


@interface FOutstandingPut : NSObject

@property (nonatomic, strong) NSString *action;
@property (nonatomic, strong) NSDictionary *request;
@property (nonatomic, copy) fbt_void_nsstring_nsstring onCompleteBlock;
@property (nonatomic) BOOL sent;

@end

@implementation FOutstandingPut

@end


typedef enum {
    ConnectionStateDisconnected,
    ConnectionStateGettingToken,
    ConnectionStateConnecting,
    ConnectionStateAuthenticating,
    ConnectionStateConnected
} ConnectionState;

@interface FPersistentConnection () {
    ConnectionState connectionState;
    BOOL firstConnection;
    NSTimeInterval reconnectDelay;
    NSTimeInterval lastConnectionAttemptTime;
    NSTimeInterval lastConnectionEstablishedTime;
    SCNetworkReachabilityRef reachability;
}

- (int) getNextRequestNumber;
- (void) onDataPushWithAction:(NSString *)action andBody:(NSDictionary *)body;
- (void) handleTimestamp:(NSNumber *)timestamp;
- (void) sendOnDisconnectAction:(NSString *)action forPath:(NSString *)pathString withData:(id)data andCallback:(fbt_void_nsstring_nsstring)callback;

@property (nonatomic, strong) FConnection* realtime;
@property (nonatomic, strong) NSMutableDictionary* listens;
@property (nonatomic, strong) NSMutableDictionary* outstandingPuts;
@property (nonatomic, strong) NSMutableArray* onDisconnectQueue;
@property (nonatomic, strong) FRepoInfo* repoInfo;
@property (nonatomic, strong) FAtomicNumber* putCounter;
@property (nonatomic, strong) FAtomicNumber* requestNumber;
@property (nonatomic, strong) NSMutableDictionary* requestCBHash;
@property (nonatomic, strong) FIRDatabaseConfig *config;
@property (nonatomic) NSUInteger unackedListensCount;
@property (nonatomic, strong) NSMutableArray *putsToAck;
@property (nonatomic, strong) dispatch_queue_t dispatchQueue;
@property (nonatomic, strong) NSString* lastSessionID;
@property (nonatomic, strong) NSMutableSet *interruptReasons;
@property (nonatomic, strong) FIRRetryHelper *retryHelper;
@property (nonatomic, strong) id<FAuthTokenProvider> authTokenProvider;
@property (nonatomic, strong) NSString *authToken;
@property (nonatomic) BOOL forceAuthTokenRefresh;
@property (nonatomic) NSUInteger currentFetchTokenAttempt;

@end


@implementation FPersistentConnection

- (id)initWithRepoInfo:(FRepoInfo *)repoInfo dispatchQueue:(dispatch_queue_t)dispatchQueue config:(FIRDatabaseConfig *)config {
    self = [super init];
    if (self) {
        self->_config = config;
        self->_repoInfo = repoInfo;
        self->_dispatchQueue = dispatchQueue;
        self->_authTokenProvider = config.authTokenProvider;
        NSAssert(self->_authTokenProvider != nil, @"Expected auth token provider");
        self.interruptReasons = [NSMutableSet set];

        self.listens = [[NSMutableDictionary alloc] init];
        self.outstandingPuts = [[NSMutableDictionary alloc] init];
        self.onDisconnectQueue = [[NSMutableArray alloc] init];
        self.putCounter = [[FAtomicNumber alloc] init];
        self.requestNumber = [[FAtomicNumber alloc] init];
        self.requestCBHash = [[NSMutableDictionary alloc] init];
        self.unackedListensCount = 0;
        self.putsToAck = [NSMutableArray array];
        connectionState = ConnectionStateDisconnected;
        firstConnection = YES;
        reconnectDelay = kPersistentConnReconnectMinDelay;

        self->_retryHelper = [[FIRRetryHelper alloc] initWithDispatchQueue:dispatchQueue
                                                 minRetryDelayAfterFailure:kPersistentConnReconnectMinDelay
                                                             maxRetryDelay:kPersistentConnReconnectMaxDelay
                                                             retryExponent:kPersistentConnReconnectMultiplier
                                                              jitterFactor:0.7];

        [self setupNotifications];
        // Make sure we don't actually connect until open is called
        [self interruptForReason:kFInterruptReasonWaitingForOpen];
    }
    // nb: The reason establishConnection isn't called here like the JS version is because
    // callers need to set the delegate first. The ctor can be modified to accept the delegate
    // but that deviates from normal ios conventions. After the delegate has been set, the caller
    // is responsible for calling establishConnection:
    return self;
}

- (void) dealloc {
    if (reachability) {
        // Unschedule the notifications
        SCNetworkReachabilitySetDispatchQueue(reachability, NULL);
        CFRelease(reachability);
    }
}

#pragma mark -
#pragma mark Public methods

- (void) open {
    [self resumeForReason:kFInterruptReasonWaitingForOpen];
}

/**
* Note that the listens dictionary has a type of Map[String (pathString), Map[FQueryParams, FOutstandingQuery]]
*
* This means, for each path we care about, there are sets of queryParams that correspond to an FOutstandingQuery object.
* There can be multiple sets at a path since we overlap listens for a short time while adding or removing a query from a
* location in the tree.
*/
- (void) listen:(FQuerySpec *)query
          tagId:(NSNumber *)tagId
           hash:(id<FSyncTreeHash>)hash
     onComplete:(fbt_void_nsstring)onComplete {
    FFLog(@"I-RDB034001", @"Listen called for %@", query);

    NSAssert(self.listens[query] == nil, @"listen() called twice for the same query");
    NSAssert(query.isDefault || !query.loadsAllData, @"listen called for non-default but complete query");
    FOutstandingQuery* outstanding = [[FOutstandingQuery alloc] init];
    outstanding.query = query;
    outstanding.tagId = tagId;
    outstanding.syncTreeHash = hash;
    outstanding.onComplete = onComplete;
    [self.listens setObject:outstanding forKey:query];
    if ([self connected]) {
        [self sendListen:outstanding];
    }
}

- (void) putData:(id)data forPath:(NSString *)pathString withHash:(NSString *)hash withCallback:(fbt_void_nsstring_nsstring)onComplete {
    [self putInternal:data forAction:kFWPRequestActionPut forPath:pathString withHash:hash withCallback:onComplete];
}

- (void) mergeData:(id)data forPath:(NSString *)pathString withCallback:(fbt_void_nsstring_nsstring)onComplete {
    [self putInternal:data forAction:kFWPRequestActionMerge forPath:pathString withHash:nil withCallback:onComplete];
}

- (void) onDisconnectPutData:(id)data forPath:(FPath *)path withCallback:(fbt_void_nsstring_nsstring)callback {
    if ([self canSendWrites]) {
        [self sendOnDisconnectAction:kFWPRequestActionDisconnectPut forPath:[path description] withData:data andCallback:callback];
    } else {
        FTupleOnDisconnect* tuple = [[FTupleOnDisconnect alloc] init];
        tuple.pathString = [path description];
        tuple.action = kFWPRequestActionDisconnectPut;
        tuple.data = data;
        tuple.onComplete = callback;
        [self.onDisconnectQueue addObject:tuple];
    }
}

- (void) onDisconnectMergeData:(id)data forPath:(FPath *)path withCallback:(fbt_void_nsstring_nsstring)callback {
    if ([self canSendWrites]) {
        [self sendOnDisconnectAction:kFWPRequestActionDisconnectMerge forPath:[path description] withData:data andCallback:callback];
    } else {
        FTupleOnDisconnect* tuple = [[FTupleOnDisconnect alloc] init];
        tuple.pathString = [path description];
        tuple.action = kFWPRequestActionDisconnectMerge;
        tuple.data = data;
        tuple.onComplete = callback;
        [self.onDisconnectQueue addObject:tuple];
    }
}

- (void) onDisconnectCancelPath:(FPath *)path withCallback:(fbt_void_nsstring_nsstring)callback {
    if ([self canSendWrites]) {
        [self sendOnDisconnectAction:kFWPRequestActionDisconnectCancel forPath:[path description] withData:[NSNull null] andCallback:callback];
    } else {
        FTupleOnDisconnect* tuple = [[FTupleOnDisconnect alloc] init];
        tuple.pathString = [path description];
        tuple.action = kFWPRequestActionDisconnectCancel;
        tuple.data = [NSNull null];
        tuple.onComplete = callback;
        [self.onDisconnectQueue addObject:tuple];
    }
}

- (void) unlisten:(FQuerySpec *)query tagId:(NSNumber *)tagId {
    FPath *path = query.path;
    FFLog(@"I-RDB034002", @"Unlistening for %@", query);

    NSArray *outstanding = [self removeListen:query];
    if (outstanding.count > 0 && [self connected]) {
        [self sendUnlisten:path queryParams:query.params tagId:tagId];
    }
}

- (void) refreshAuthToken:(NSString *)token {
    self.authToken = token;
    if ([self connected]) {
        if (token != nil) {
            [self sendAuthAndRestoreStateAfterComplete:NO];
        } else {
            [self sendUnauth];
        }
    }
}

#pragma mark -
#pragma mark Connection status

- (BOOL)connected {
    return self->connectionState == ConnectionStateAuthenticating || self->connectionState == ConnectionStateConnected;
}

- (BOOL)canSendWrites {
    return self->connectionState == ConnectionStateConnected;
}

#pragma mark -
#pragma mark FConnection delegate methods

- (void)onReady:(FConnection *)fconnection atTime:(NSNumber *)timestamp sessionID:(NSString *)sessionID {
    FFLog(@"I-RDB034003", @"On ready");
    lastConnectionEstablishedTime = [[NSDate date] timeIntervalSince1970];
    [self handleTimestamp:timestamp];

    if (firstConnection) {
        [self sendConnectStats];
    }

    [self restoreAuth];
    firstConnection = NO;
    self.lastSessionID = sessionID;
    dispatch_async(self.dispatchQueue, ^{
        [self.delegate onConnect:self];
    });
}

- (void)onDataMessage:(FConnection *)fconnection withMessage:(NSDictionary *)message {
    if (message[kFWPRequestNumber] != nil) {
        // this is a response to a request we sent
        NSNumber* rn = [NSNumber numberWithInt:[[message objectForKey:kFWPRequestNumber] intValue]];
        if ([self.requestCBHash objectForKey:rn]) {
            void (^callback)(NSDictionary*) = [self.requestCBHash objectForKey:rn];
            [self.requestCBHash removeObjectForKey:rn];

            if (callback) {
                //dispatch_async(self.dispatchQueue, ^{
                    callback([message objectForKey:kFWPResponseForRNData]);
                //});
            }
        }
    } else if (message[kFWPRequestError] != nil) {
        NSString* error = [message objectForKey:kFWPRequestError];
        @throw [[NSException alloc] initWithName:@"FirebaseDatabaseServerError" reason:error userInfo:nil];
    } else if (message[kFWPAsyncServerAction] != nil) {
        // this is a server push of some sort
        NSString* action = [message objectForKey:kFWPAsyncServerAction];
        NSDictionary* body = [message objectForKey:kFWPAsyncServerPayloadBody];
        [self onDataPushWithAction:action andBody:body];
    }
}

- (void)onDisconnect:(FConnection *)fconnection withReason:(FDisconnectReason)reason {
    FFLog(@"I-RDB034004", @"Got on disconnect due to %s", (reason == DISCONNECT_REASON_SERVER_RESET) ? "server_reset" : "other");
    connectionState = ConnectionStateDisconnected;
    // Drop the realtime connection
    self.realtime = nil;
    [self cancelSentTransactions];
    [self.requestCBHash removeAllObjects];
    self.unackedListensCount = 0;
    if ([self shouldReconnect]) {
        NSTimeInterval timeSinceLastConnectSucceeded = [[NSDate date] timeIntervalSince1970] - lastConnectionEstablishedTime;
        BOOL lastConnectionWasSuccessful;
        if (lastConnectionEstablishedTime > 0) {
            lastConnectionWasSuccessful = timeSinceLastConnectSucceeded > kPersistentConnSuccessfulConnectionEstablishedDelay;
        } else {
            lastConnectionWasSuccessful = NO;
        }

        if (reason == DISCONNECT_REASON_SERVER_RESET || lastConnectionWasSuccessful) {
            [self.retryHelper signalSuccess];
        }
        [self tryScheduleReconnect];
    }
    lastConnectionEstablishedTime = 0;
    [self.delegate onDisconnect:self];
}

- (void)onKill:(FConnection *)fconnection withReason:(NSString *)reason {
    FFWarn(@"I-RDB034005", @"Firebase Database connection was forcefully killed by the server.  Will not attempt reconnect. Reason: %@", reason);
    [self interruptForReason:kFInterruptReasonServerKill];
}

#pragma mark -
#pragma mark Connection handling methods

- (void) interruptForReason:(NSString *)reason {
    FFLog(@"I-RDB034006", @"Connection interrupted for: %@", reason);

    [self.interruptReasons addObject:reason];
    if (self.realtime) {
        // Will call onDisconnect and set the connection state to Disconnected
        [self.realtime close];
        self.realtime = nil;
    } else {
        [self.retryHelper cancel];
        self->connectionState = ConnectionStateDisconnected;
    }
    // Reset timeouts
    [self.retryHelper signalSuccess];
}

- (void) resumeForReason:(NSString *)reason {
    FFLog(@"I-RDB034007", @"Connection no longer interrupted for: %@", reason);
    [self.interruptReasons removeObject:reason];

    if ([self shouldReconnect] && connectionState == ConnectionStateDisconnected) {
        [self tryScheduleReconnect];
    }
}

- (BOOL) shouldReconnect {
    return self.interruptReasons.count == 0;
}

- (BOOL) isInterruptedForReason:(NSString *)reason {
    return [self.interruptReasons containsObject:reason];
}

#pragma mark -
#pragma mark Private methods

- (void) tryScheduleReconnect {
    if ([self shouldReconnect]) {
        NSAssert(self->connectionState == ConnectionStateDisconnected,
                 @"Not in disconnected state: %d", self->connectionState);
        BOOL forceRefresh = self.forceAuthTokenRefresh;
        self.forceAuthTokenRefresh = NO;
        FFLog(@"I-RDB034008", @"Scheduling connection attempt");
        [self.retryHelper retry:^{
            FFLog(@"I-RDB034009", @"Trying to fetch auth token");
            NSAssert(self->connectionState == ConnectionStateDisconnected,
                     @"Not in disconnected state: %d", self->connectionState);
            self->connectionState = ConnectionStateGettingToken;
            self.currentFetchTokenAttempt++;
            NSUInteger thisFetchTokenAttempt = self.currentFetchTokenAttempt;
            [self.authTokenProvider fetchTokenForcingRefresh:forceRefresh withCallback:^(NSString *token, NSError *error) {
                if (thisFetchTokenAttempt == self.currentFetchTokenAttempt) {
                    if (error != nil) {
                        self->connectionState = ConnectionStateDisconnected;
                        FFLog(@"I-RDB034010", @"Error fetching token: %@", error);
                        [self tryScheduleReconnect];
                    } else {
                        // Someone could have interrupted us while fetching the token,
                        // marking the connection as Disconnected
                        if (self->connectionState == ConnectionStateGettingToken) {
                            FFLog(@"I-RDB034011", @"Successfully fetched token, opening connection");
                            [self openNetworkConnectionWithToken:token];
                        } else {
                            NSAssert(self->connectionState == ConnectionStateDisconnected,
                                     @"Expected connection state disconnected, but got %d", self->connectionState);
                            FFLog(@"I-RDB034012", @"Not opening connection after token refresh, because connection was set to disconnected.");
                        }
                    }
                } else {
                    FFLog(@"I-RDB034013", @"Ignoring fetch token result, because this was not the latest attempt.");
                }
            }];
        }];

    }
}

- (void) openNetworkConnectionWithToken:(NSString *)token {
    NSAssert(self->connectionState == ConnectionStateGettingToken,
             @"Trying to open network connection while in wrong state: %d", self->connectionState);
    self.authToken = token;
    self->connectionState = ConnectionStateConnecting;
    self.realtime = [[FConnection alloc] initWith:self.repoInfo
                                 andDispatchQueue:self.dispatchQueue
                                    lastSessionID:self.lastSessionID];
    self.realtime.delegate = self;
    [self.realtime open];
}

static void reachabilityCallback(SCNetworkReachabilityRef ref, SCNetworkReachabilityFlags flags, void* info) {
    if (flags & kSCNetworkReachabilityFlagsReachable) {
        FFLog(@"I-RDB034014", @"Network became reachable. Trigger a connection attempt");
        FPersistentConnection* self = (__bridge FPersistentConnection *)info;
        // Reset reconnect delay
        [self.retryHelper signalSuccess];
        if (self->connectionState == ConnectionStateDisconnected) {
            [self tryScheduleReconnect];
        }
    } else {
        FFLog(@"I-RDB034015", @"Network is not reachable");
    }
}

- (void) enteringForeground {
    dispatch_async(self.dispatchQueue, ^{
        // Reset reconnect delay
        [self.retryHelper signalSuccess];
        if (self->connectionState == ConnectionStateDisconnected) {
            [self tryScheduleReconnect];
        }
    });
}

- (void) setupNotifications {

    NSString * const* foregroundConstant = (NSString * const *) dlsym(RTLD_DEFAULT, "UIApplicationWillEnterForegroundNotification");
    if (foregroundConstant) {
        [[NSNotificationCenter defaultCenter] addObserver:self
                                                 selector:@selector(enteringForeground)
                                                     name:*foregroundConstant
                                                   object:nil];
    }
    // An empty address is interpreted a generic internet access
    struct sockaddr_in zeroAddress;
    bzero(&zeroAddress, sizeof(zeroAddress));
    zeroAddress.sin_len = sizeof(zeroAddress);
    zeroAddress.sin_family = AF_INET;
    reachability = SCNetworkReachabilityCreateWithAddress(kCFAllocatorDefault, (const struct sockaddr *)&zeroAddress);
    SCNetworkReachabilityContext ctx = {0, (__bridge void *)(self), NULL, NULL, NULL};
    if (SCNetworkReachabilitySetCallback(reachability, reachabilityCallback, &ctx)) {
        SCNetworkReachabilitySetDispatchQueue(reachability, self.dispatchQueue);
    } else {
        FFLog(@"I-RDB034016", @"Failed to set up network reachability monitoring");
        CFRelease(reachability);
        reachability = NULL;
    }
}

- (void) sendAuthAndRestoreStateAfterComplete:(BOOL)restoreStateAfterComplete {
    NSAssert([self connected], @"Must be connected to send auth");
    NSAssert(self.authToken != nil, @"Can't send auth if there is no credential");

    NSDictionary* requestData = @{kFWPRequestCredential: self.authToken};
    [self sendAction:kFWPRequestActionAuth body:requestData sensitive:YES callback:^(NSDictionary *data) {
        self->connectionState = ConnectionStateConnected;
        NSString* status = [data objectForKey:kFWPResponseForActionStatus];
        id responseData = [data objectForKey:kFWPResponseForActionData];
        if (responseData == nil) {
            responseData = @"error";
        }

        BOOL statusOk = [status isEqualToString:kFWPResponseForActionStatusOk];
        if (statusOk) {
            if (restoreStateAfterComplete) {
                [self restoreState];
            }
        } else {
            self.authToken = nil;
            self.forceAuthTokenRefresh = YES;
            if ([status isEqualToString:@"expired_token"]) {
                FFLog(@"I-RDB034017", @"Authentication failed: %@ (%@)", status, responseData);
            } else {
                FFWarn(@"I-RDB034018", @"Authentication failed: %@ (%@)", status, responseData);
            }
            [self.realtime close];
        }
    }];
}

- (void) sendUnauth {
    [self sendAction:kFWPRequestActionUnauth body:@{} sensitive:NO callback:nil];
}

- (void) onAuthRevokedWithStatus:(NSString *)status andReason:(NSString *)reason {
    // This might be for an earlier token than we just recently sent. But since we need to close the connection anyways,
    // we can set it to null here and we will refresh the token later on reconnect
    if ([status isEqualToString:@"expired_token"]) {
        FFLog(@"I-RDB034019", @"Auth token revoked: %@ (%@)", status, reason);
    } else {
        FFWarn(@"I-RDB034020", @"Auth token revoked: %@ (%@)", status, reason);
    }
    self.authToken = nil;
    self.forceAuthTokenRefresh = YES;
    // Try reconnecting on auth revocation
    [self.realtime close];
}

- (void) onListenRevoked:(FPath *)path {
    NSArray *queries = [self removeAllListensAtPath:path];
    for (FOutstandingQuery* query in queries) {
        query.onComplete(@"permission_denied");
    }
}

- (void) sendOnDisconnectAction:(NSString *)action forPath:(NSString *)pathString withData:(id)data andCallback:(fbt_void_nsstring_nsstring)callback {

    NSDictionary* request = @{kFWPRequestPath: pathString, kFWPRequestData: data};
    FFLog(@"I-RDB034021", @"onDisconnect %@: %@", action, request);

    [self sendAction:action
                body:request
           sensitive:NO
            callback:^(NSDictionary *data) {
            NSString* status = [data objectForKey:kFWPResponseForActionStatus];
            NSString* errorReason = [data objectForKey:kFWPResponseForActionData];
            callback(status, errorReason);
        }];
}

- (void) sendPut:(NSNumber *) index {
    NSAssert([self canSendWrites], @"sendPut called when not able to send writes");
    FOutstandingPut* put = self.outstandingPuts[index];
    assert(put != nil);
    fbt_void_nsstring_nsstring onComplete = put.onCompleteBlock;

    // Do not async this block; copying the block insinde sendAction: doesn't happen in time (or something) so coredumps
    put.sent = YES;
    [self sendAction:put.action
                body:put.request
           sensitive:NO
            callback:^(NSDictionary* data) {

            FOutstandingPut *currentPut = self.outstandingPuts[index];
            if (currentPut == put) {
                [self.outstandingPuts removeObjectForKey:index];

                if (onComplete != nil) {
                    NSString *status = [data objectForKey:kFWPResponseForActionStatus];
                    NSString *errorReason = [data objectForKey:kFWPResponseForActionData];
                    if (self.unackedListensCount == 0) {
                        onComplete(status, errorReason);
                    } else {
                        FTupleCallbackStatus *putToAck = [[FTupleCallbackStatus alloc] init];
                        putToAck.block = onComplete;
                        putToAck.status = status;
                        putToAck.errorReason = errorReason;
                        [self.putsToAck addObject:putToAck];
                    }
                }
            } else {
                FFLog(@"I-RDB034022", @"Ignoring on complete for put %@ because it was already removed", index);
            }
        }];
}

- (void) sendUnlisten:(FPath *)path queryParams:(FQueryParams *)queryParams tagId:(NSNumber *)tagId {
    FFLog(@"I-RDB034023", @"Unlisten on %@ for %@", path, queryParams);

    NSMutableDictionary* request = [NSMutableDictionary dictionaryWithObjectsAndKeys:[path toString], kFWPRequestPath, nil];
    if (tagId != nil) {
        [request setObject:queryParams.wireProtocolParams forKey:kFWPRequestQueries];
        [request setObject:tagId forKey:kFWPRequestTag];
    }

    [self sendAction:kFWPRequestActionTaggedUnlisten
                body:request
           sensitive:NO
            callback:nil];
}

- (void) putInternal:(id)data forAction:(NSString *)action forPath:(NSString *)pathString withHash:(NSString *)hash withCallback:(fbt_void_nsstring_nsstring)onComplete {

    NSMutableDictionary *request = [NSMutableDictionary dictionaryWithObjectsAndKeys:
                                    pathString, kFWPRequestPath,
                                    data, kFWPRequestData, nil];
    if(hash) {
        [request setObject:hash forKey:kFWPRequestHash];
    }

    FOutstandingPut *put = [[FOutstandingPut alloc] init];
    put.action = action;
    put.request = request;
    put.onCompleteBlock = onComplete;
    put.sent = NO;

    NSNumber* index = [self.putCounter getAndIncrement];
    self.outstandingPuts[index] = put;

    if ([self canSendWrites]) {
        FFLog(@"I-RDB034024", @"Was connected, and added as index: %@", index);
        [self sendPut:index];
    }
    else {
        FFLog(@"I-RDB034025", @"Wasn't connected or writes paused, so added to outstanding puts only. Path: %@", pathString);
    }
}

- (void) sendListen:(FOutstandingQuery *)listenSpec {
    FQuerySpec *query = listenSpec.query;
    FFLog(@"I-RDB034026", @"Listen for %@", query);
    NSMutableDictionary *request = [NSMutableDictionary dictionaryWithObject:[query.path toString] forKey:kFWPRequestPath];

    // Only bother to send query if it's non-default
    if (listenSpec.tagId != nil) {
        [request setObject:[query.params wireProtocolParams] forKey:kFWPRequestQueries];
        [request setObject:listenSpec.tagId forKey:kFWPRequestTag];
    }

    [request setObject:[listenSpec.syncTreeHash simpleHash] forKey:kFWPRequestHash];
    if ([listenSpec.syncTreeHash includeCompoundHash]) {
        FCompoundHash *compoundHash = [listenSpec.syncTreeHash compoundHash];
        NSMutableArray *posts = [NSMutableArray array];
        for (FPath *path in compoundHash.posts) {
            [posts addObject:path.wireFormat];
        }
        request[kFWPRequestCompoundHash] = @{ kFWPRequestCompoundHashHashes: compoundHash.hashes,
                                              kFWPRequestCompoundHashPaths: posts };
    }

    fbt_void_nsdictionary onResponse = ^(NSDictionary *response) {
        FFLog(@"I-RDB034027", @"Listen response %@", response);
        // warn in any case, even if the listener was removed
        [self warnOnListenWarningsForQuery:query payload:response[kFWPResponseForActionData]];

        FOutstandingQuery *currentListenSpec = self.listens[query];

        // only trigger actions if the listen hasn't been removed (and maybe readded)
        if (currentListenSpec == listenSpec) {
            NSString *status = [response objectForKey:kFWPRequestStatus];
            if (![status isEqualToString:@"ok"]) {
                [self removeListen:query];
            }

            if (listenSpec.onComplete) {
                listenSpec.onComplete(status);
            }
        }

        self.unackedListensCount--;
        NSAssert(self.unackedListensCount >= 0, @"unackedListensCount decremented to be negative.");
        if (self.unackedListensCount == 0) {
            [self ackPuts];
        }
    };

    [self sendAction:kFWPRequestActionTaggedListen
                body:request
           sensitive:NO
            callback:onResponse];

    self.unackedListensCount++;
}

- (void) warnOnListenWarningsForQuery:(FQuerySpec *)query payload:(id)payload {
    if (payload != nil && [payload isKindOfClass:[NSDictionary class]]) {
        NSDictionary *payloadDict = payload;
        id warnings = payloadDict[kFWPResponseDataWarnings];
        if (warnings != nil && [warnings isKindOfClass:[NSArray class]]) {
            NSArray *warningsArr = warnings;
            if ([warningsArr containsObject:@"no_index"]) {
                NSString *indexSpec = [NSString stringWithFormat:@"\".indexOn\": \"%@\"", [query.params.index queryDefinition]];
                NSString *indexPath = [query.path description];
                FFWarn(@"I-RDB034028", @"Using an unspecified index. Your data will be downloaded and filtered on the client. "
                       "Consider adding %@ at %@ to your security rules for better performance", indexSpec, indexPath);
            }
        }
    }
}

- (int) getNextRequestNumber {
    return [[self.requestNumber getAndIncrement] intValue];
}

- (void)sendAction:(NSString *)action
              body:(NSDictionary *)message
         sensitive:(BOOL)sensitive
          callback:(void (^)(NSDictionary* data))onMessage {
    // Hold onto the onMessage callback for this request before firing it off
    NSNumber* rn = [NSNumber numberWithInt:[self getNextRequestNumber]];
    NSDictionary* msg = [NSDictionary dictionaryWithObjectsAndKeys:
                                    rn, kFWPRequestNumber,
                                    action, kFWPRequestAction,
                                    message, kFWPRequestPayloadBody,
                         nil];

    [self.realtime sendRequest:msg sensitive:sensitive];

    if (onMessage) {
        // Debug message without a callback; bump the rn, but don't hold onto the cb
        [self.requestCBHash setObject:[onMessage copy] forKey:rn];
    }
}

- (void) cancelSentTransactions {
    NSMutableDictionary<NSNumber*, FOutstandingPut*>* cancelledOutstandingPuts =  [[NSMutableDictionary alloc] init];

    for (NSNumber* index in self.outstandingPuts) {
        FOutstandingPut* put = self.outstandingPuts[index];
        if (put.request[kFWPRequestHash] && put.sent) {
            // This is a sent transaction put.
            cancelledOutstandingPuts[index] = put;
        }
    }

    [cancelledOutstandingPuts enumerateKeysAndObjectsUsingBlock:^(NSNumber *index, FOutstandingPut *outstandingPut, BOOL *stop) {
        // `onCompleteBlock:` may invoke `rerunTransactionsForPath:` and enqueue new writes. We defer calling
        // it until we have finished enumerating all existing writes.
        outstandingPut.onCompleteBlock(kFTransactionDisconnect, @"Client was disconnected while running a transaction");
        [self.outstandingPuts removeObjectForKey:index];
    }];
}

- (void) onDataPushWithAction:(NSString *)action andBody:(NSDictionary *)body {
    FFLog(@"I-RDB034029", @"handleServerMessage: %@, %@", action, body);
    id<FPersistentConnectionDelegate> delegate = self.delegate;
    if ([action isEqualToString:kFWPAsyncServerDataUpdate] || [action isEqualToString:kFWPAsyncServerDataMerge]) {
        BOOL isMerge = [action isEqualToString:kFWPAsyncServerDataMerge];

        if ([body objectForKey:kFWPAsyncServerDataUpdateBodyPath] && [body objectForKey:kFWPAsyncServerDataUpdateBodyData]) {
            NSString* path = [body objectForKey:kFWPAsyncServerDataUpdateBodyPath];
            id payloadData = [body objectForKey:kFWPAsyncServerDataUpdateBodyData];
            if (isMerge && [payloadData isKindOfClass:[NSDictionary class]] && [payloadData count] == 0) {
                // ignore empty merge
            } else {
                [delegate onDataUpdate:self forPath:path message:payloadData isMerge:isMerge tagId:[body objectForKey:kFWPAsyncServerDataUpdateBodyTag]];
            }
        }
        else {
            FFLog(@"I-RDB034030", @"Malformed data response from server missing path or data: %@", body);
        }
    } else if ([action isEqualToString:kFWPAsyncServerDataRangeMerge]) {
        NSString *path = body[kFWPAsyncServerDataUpdateBodyPath];
        NSArray *ranges = body[kFWPAsyncServerDataUpdateBodyData];
        NSNumber *tag = body[kFWPAsyncServerDataUpdateBodyTag];
        NSMutableArray *rangeMerges = [NSMutableArray array];
        for (NSDictionary *range in ranges) {
            NSString *startString = range[kFWPAsyncServerDataUpdateStartPath];
            NSString *endString = range[kFWPAsyncServerDataUpdateEndPath];
            id updateData = range[kFWPAsyncServerDataUpdateRangeMerge];
            id<FNode> updates = [FSnapshotUtilities nodeFrom:updateData];
            FPath *start = (startString != nil) ? [[FPath alloc] initWith:startString] : nil;
            FPath *end = (endString != nil) ? [[FPath alloc] initWith:endString] : nil;
            FRangeMerge *merge = [[FRangeMerge alloc] initWithStart:start end:end updates:updates];
            [rangeMerges addObject:merge];
        }
        [delegate onRangeMerge:rangeMerges forPath:path tagId:tag];
    } else if ([action isEqualToString:kFWPAsyncServerAuthRevoked]) {
        NSString* status = [body objectForKey:kFWPResponseForActionStatus];
        NSString* reason = [body objectForKey:kFWPResponseForActionData];
        [self onAuthRevokedWithStatus:status andReason:reason];
    } else if ([action isEqualToString:kFWPASyncServerListenCancelled]) {
        NSString* pathString = [body objectForKey:kFWPAsyncServerDataUpdateBodyPath];
        [self onListenRevoked:[[FPath alloc] initWith:pathString]];
    } else if ([action isEqualToString:kFWPAsyncServerSecurityDebug]) {
        NSString* msg = [body objectForKey:@"msg"];
        if (msg != nil) {
            NSArray *msgs = [msg componentsSeparatedByString:@"\n"];
            for (NSString* m in msgs) {
                FFWarn(@"I-RDB034031", @"%@", m);
            }
        }
    } else {
        // TODO: revoke listens, auth, security debug
        FFLog(@"I-RDB034032", @"Unsupported action from server: %@", action);
    }
}

- (void) restoreAuth {
    FFLog(@"I-RDB034033", @"Calling restore state");

    NSAssert(self->connectionState == ConnectionStateConnecting,
             @"Wanted to restore auth, but was in wrong state: %d", self->connectionState);
    if (self.authToken == nil) {
        FFLog(@"I-RDB034034", @"Not restoring auth because token is nil");
        self->connectionState = ConnectionStateConnected;
        [self restoreState];
    } else {
        FFLog(@"I-RDB034035", @"Restoring auth");
        self->connectionState = ConnectionStateAuthenticating;
        [self sendAuthAndRestoreStateAfterComplete:YES];
    }
}

- (void) restoreState {
    NSAssert(self->connectionState == ConnectionStateConnected,
             @"Should be connected if we're restoring state, but we are: %d", self->connectionState);

    [self.listens enumerateKeysAndObjectsUsingBlock:^(FQuerySpec *query, FOutstandingQuery *outstandingListen, BOOL *stop) {
        FFLog(@"I-RDB034036", @"Restoring listen for %@", query);
        [self sendListen:outstandingListen];
    }];

    NSArray* keys = [[self.outstandingPuts allKeys] sortedArrayUsingSelector:@selector(compare:)];
    for(int i = 0; i < [keys count]; i++) {
        if([self.outstandingPuts objectForKey:[keys objectAtIndex:i]] != nil) {
            FFLog(@"I-RDB034037", @"Restoring put: %d", i);
            [self sendPut:[keys objectAtIndex:i]];
        }
        else {
            FFLog(@"I-RDB034038", @"Restoring put: skipped nil: %d", i);
        }
    }

    for (FTupleOnDisconnect* tuple in self.onDisconnectQueue) {
        [self sendOnDisconnectAction:tuple.action forPath:tuple.pathString withData:tuple.data andCallback:tuple.onComplete];
    }
    [self.onDisconnectQueue removeAllObjects];
}

- (NSArray *) removeListen:(FQuerySpec *)query {
    NSAssert(query.isDefault || !query.loadsAllData, @"removeListen called for non-default but complete query");

    FOutstandingQuery* outstanding = self.listens[query];
    if (!outstanding) {
        FFLog(@"I-RDB034039", @"Trying to remove listener for query %@ but no listener exists", query);
        return @[];
    } else {
        [self.listens removeObjectForKey:query];
        return @[outstanding];
    }
}

- (NSArray *) removeAllListensAtPath:(FPath *)path {
    FFLog(@"I-RDB034040", @"Removing all listens at path %@", path);
    NSMutableArray *removed = [NSMutableArray array];
    NSMutableArray *toRemove = [NSMutableArray array];
    [self.listens enumerateKeysAndObjectsUsingBlock:^(FQuerySpec *spec, FOutstandingQuery *outstanding, BOOL *stop) {
        if ([spec.path isEqual:path]) {
            [removed addObject:outstanding];
            [toRemove addObject:spec];
        }
    }];
    [self.listens removeObjectsForKeys:toRemove];
    return removed;
}

- (void) purgeOutstandingWrites {
    // We might have unacked puts in our queue that we need to ack now before we send out any cancels...
    [self ackPuts];
    // Cancel in order
    NSArray* keys = [[self.outstandingPuts allKeys] sortedArrayUsingSelector:@selector(compare:)];
    for (NSNumber *key in keys) {
        FOutstandingPut *put = self.outstandingPuts[key];
        if (put.onCompleteBlock != nil) {
            put.onCompleteBlock(kFErrorWriteCanceled, nil);
        }
    }
    for (FTupleOnDisconnect *onDisconnect in self.onDisconnectQueue) {
        if (onDisconnect.onComplete != nil) {
            onDisconnect.onComplete(kFErrorWriteCanceled, nil);
        }
    }
    [self.outstandingPuts removeAllObjects];
    [self.onDisconnectQueue removeAllObjects];
}

- (void) ackPuts {
    for (FTupleCallbackStatus *put in self.putsToAck) {
        put.block(put.status, put.errorReason);
    }
    [self.putsToAck removeAllObjects];
}

- (void) handleTimestamp:(NSNumber *)timestamp {
    FFLog(@"I-RDB034041", @"Handling timestamp: %@", timestamp);
    double timestampDeltaMs = [timestamp doubleValue] - ([[NSDate date] timeIntervalSince1970] * 1000);
    [self.delegate onServerInfoUpdate:self updates:@{kDotInfoServerTimeOffset: [NSNumber numberWithDouble:timestampDeltaMs]}];
}

- (void) sendStats:(NSDictionary *)stats {
    if ([stats count] > 0) {
        NSDictionary *request = @{ kFWPRequestCounters: stats };
        [self sendAction:kFWPRequestActionStats body:request sensitive:NO callback:^(NSDictionary *data) {
            NSString* status = [data objectForKey:kFWPResponseForActionStatus];
            NSString* errorReason = [data objectForKey:kFWPResponseForActionData];
            BOOL statusOk = [status isEqualToString:kFWPResponseForActionStatusOk];
            if (!statusOk) {
                FFLog(@"I-RDB034042", @"Failed to send stats: %@", errorReason);
            }
        }];
    } else {
        FFLog(@"I-RDB034043", @"Not sending stats because stats are empty");
    }
}

- (void) sendConnectStats {
    NSMutableDictionary *stats = [NSMutableDictionary dictionary];

    #if TARGET_OS_IOS || TARGET_OS_TV
    if (self.config.persistenceEnabled) {
        stats[@"persistence.ios.enabled"] = @1;
    }
    #elif TARGET_OS_OSX
    if (self.config.persistenceEnabled) {
        stats[@"persistence.osx.enabled"] = @1;
    }
    #endif
    NSString *sdkVersion = [[FIRDatabase sdkVersion] stringByReplacingOccurrencesOfString:@"." withString:@"-"];
    NSString *sdkStatName = [NSString stringWithFormat:@"sdk.objc.%@", sdkVersion];
    stats[sdkStatName] = @1;
    FFLog(@"I-RDB034044", @"Sending first connection stats");
    [self sendStats:stats];
}

- (NSDictionary *) dumpListens {
    return self.listens;
}

@end