diff --git a/android/src/main/AndroidManifest.xml b/android/src/main/AndroidManifest.xml index 680a42a..bd83d3b 100644 --- a/android/src/main/AndroidManifest.xml +++ b/android/src/main/AndroidManifest.xml @@ -1,3 +1,4 @@ + diff --git a/android/src/main/java/com/metarouter/reactnative/MetaRouterNetworkMonitorModule.kt b/android/src/main/java/com/metarouter/reactnative/MetaRouterNetworkMonitorModule.kt new file mode 100644 index 0000000..daaf116 --- /dev/null +++ b/android/src/main/java/com/metarouter/reactnative/MetaRouterNetworkMonitorModule.kt @@ -0,0 +1,119 @@ +package com.metarouter.reactnative + +import android.net.ConnectivityManager +import android.net.Network +import android.net.NetworkCapabilities +import android.net.NetworkRequest +import android.content.Context +import com.facebook.react.bridge.ReactApplicationContext +import com.facebook.react.bridge.ReactContextBaseJavaModule +import com.facebook.react.bridge.ReactMethod +import com.facebook.react.bridge.Promise +import com.facebook.react.bridge.Arguments +import com.facebook.react.modules.core.DeviceEventManagerModule + +class MetaRouterNetworkMonitorModule( + private val reactContext: ReactApplicationContext +) : ReactContextBaseJavaModule(reactContext) { + + override fun getName(): String = "MetaRouterNetworkMonitor" + + @Volatile + private var isConnected: Boolean = true + + private var connectivityManager: ConnectivityManager? = null + private var networkCallback: ConnectivityManager.NetworkCallback? = null + + init { + connectivityManager = try { + reactContext.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager + } catch (_: Exception) { + null + } + + // Snapshot initial state + connectivityManager?.let { cm -> + isConnected = try { + val network = cm.activeNetwork + val caps = network?.let { cm.getNetworkCapabilities(it) } + caps?.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) == true + } catch (_: Exception) { + true // fallback + } + } + + // Register callback for changes + connectivityManager?.let { cm -> + val request = NetworkRequest.Builder() + .addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET) + .build() + + val cb = object : ConnectivityManager.NetworkCallback() { + override fun onAvailable(network: Network) { + if (!isConnected) { + isConnected = true + sendEvent(true) + } + } + + override fun onLost(network: Network) { + val activeNet = cm.activeNetwork + val caps = activeNet?.let { cm.getNetworkCapabilities(it) } + val hasInternet = caps?.hasCapability( + NetworkCapabilities.NET_CAPABILITY_INTERNET + ) == true + if (!hasInternet && isConnected) { + isConnected = false + sendEvent(false) + } + } + } + + try { + cm.registerNetworkCallback(request, cb) + networkCallback = cb + } catch (_: SecurityException) { + // Missing ACCESS_NETWORK_STATE — fallback to always connected + } + } + } + + private fun sendEvent(connected: Boolean) { + val params = Arguments.createMap().apply { + putBoolean("isConnected", connected) + } + try { + reactContext + .getJSModule(DeviceEventManagerModule.RCTDeviceEventEmitter::class.java) + .emit("onConnectivityChange", params) + } catch (_: Exception) { + // JS runtime not ready yet — ignore + } + } + + @ReactMethod + fun getCurrentStatus(promise: Promise) { + promise.resolve(isConnected) + } + + @ReactMethod + fun addListener(@Suppress("UNUSED_PARAMETER") eventName: String) { + // Required for RN event emitter + } + + @ReactMethod + fun removeListeners(@Suppress("UNUSED_PARAMETER") count: Int) { + // Required for RN event emitter + } + + override fun invalidate() { + networkCallback?.let { cb -> + try { + connectivityManager?.unregisterNetworkCallback(cb) + } catch (_: Exception) { + // ignore + } + } + networkCallback = null + } +} diff --git a/android/src/main/java/com/metarouter/reactnative/MetaRouterQueueStoragePackage.kt b/android/src/main/java/com/metarouter/reactnative/MetaRouterPackage.kt similarity index 72% rename from android/src/main/java/com/metarouter/reactnative/MetaRouterQueueStoragePackage.kt rename to android/src/main/java/com/metarouter/reactnative/MetaRouterPackage.kt index c0f0f8f..13acd27 100644 --- a/android/src/main/java/com/metarouter/reactnative/MetaRouterQueueStoragePackage.kt +++ b/android/src/main/java/com/metarouter/reactnative/MetaRouterPackage.kt @@ -5,11 +5,14 @@ import com.facebook.react.bridge.NativeModule import com.facebook.react.bridge.ReactApplicationContext import com.facebook.react.uimanager.ViewManager -class MetaRouterQueueStoragePackage : ReactPackage { +class MetaRouterPackage : ReactPackage { override fun createNativeModules( reactContext: ReactApplicationContext ): List { - return listOf(MetaRouterQueueStorageModule(reactContext)) + return listOf( + MetaRouterQueueStorageModule(reactContext), + MetaRouterNetworkMonitorModule(reactContext) + ) } override fun createViewManagers( diff --git a/ios/MetaRouterNetworkMonitor.m b/ios/MetaRouterNetworkMonitor.m new file mode 100644 index 0000000..114c601 --- /dev/null +++ b/ios/MetaRouterNetworkMonitor.m @@ -0,0 +1,76 @@ +#import +#import +#import + +@interface MetaRouterNetworkMonitor : RCTEventEmitter +@end + +@implementation MetaRouterNetworkMonitor { + nw_path_monitor_t _monitor; + dispatch_queue_t _monitorQueue; + BOOL _isConnected; + BOOL _hasListeners; +} + +RCT_EXPORT_MODULE() + ++ (BOOL)requiresMainQueueSetup { + return NO; +} + +- (instancetype)init { + self = [super init]; + if (self) { + _isConnected = YES; // optimistic default + _hasListeners = NO; + _monitorQueue = dispatch_queue_create("com.metarouter.network.monitor", DISPATCH_QUEUE_SERIAL); + _monitor = nw_path_monitor_create(); + + __weak typeof(self) weakSelf = self; + nw_path_monitor_set_update_handler(_monitor, ^(nw_path_t path) { + __strong typeof(weakSelf) strongSelf = weakSelf; + if (!strongSelf) return; + + BOOL connected = (nw_path_get_status(path) == nw_path_status_satisfied); + BOOL changed = (connected != strongSelf->_isConnected); + strongSelf->_isConnected = connected; + + if (changed && strongSelf->_hasListeners) { + [strongSelf sendEventWithName:@"onConnectivityChange" + body:@{@"isConnected": @(connected)}]; + } + }); + + nw_path_monitor_set_queue(_monitor, _monitorQueue); + nw_path_monitor_start(_monitor); + } + return self; +} + +- (NSArray *)supportedEvents { + return @[@"onConnectivityChange"]; +} + +- (void)startObserving { + _hasListeners = YES; +} + +- (void)stopObserving { + _hasListeners = NO; +} + +RCT_EXPORT_METHOD(getCurrentStatus:(RCTPromiseResolveBlock)resolve + rejecter:(RCTPromiseRejectBlock)reject) +{ + dispatch_async(_monitorQueue, ^{ + resolve(@(self->_isConnected)); + }); +} + +- (void)dealloc { + if (_monitor) { + nw_path_monitor_cancel(_monitor); + } +} + +@end diff --git a/react-native.config.js b/react-native.config.js index 5a7e191..991146a 100644 --- a/react-native.config.js +++ b/react-native.config.js @@ -7,8 +7,8 @@ module.exports = { android: { sourceDir: './android', packageImportPath: - 'import com.metarouter.reactnative.MetaRouterQueueStoragePackage;', - packageInstance: 'new MetaRouterQueueStoragePackage()', + 'import com.metarouter.reactnative.MetaRouterPackage;', + packageInstance: 'new MetaRouterPackage()', }, }, }, diff --git a/src/analytics/MetaRouterAnalyticsClient.test.ts b/src/analytics/MetaRouterAnalyticsClient.test.ts index 6f5edc3..472434d 100644 --- a/src/analytics/MetaRouterAnalyticsClient.test.ts +++ b/src/analytics/MetaRouterAnalyticsClient.test.ts @@ -1,6 +1,7 @@ import { MetaRouterAnalyticsClient } from './MetaRouterAnalyticsClient'; import type { InitOptions } from './types'; import { AppState } from 'react-native'; +import { StubNetworkMonitor } from './utils/networkMonitor'; const mockAddEventListener = jest.fn(); jest @@ -594,6 +595,87 @@ describe('MetaRouterAnalyticsClient', () => { expect(client.queue[0].context.device.advertisingId).toBeUndefined(); }); + describe('network awareness', () => { + it('getDebugInfo includes networkStatus', async () => { + const monitor = new StubNetworkMonitor('connected'); + const client = new MetaRouterAnalyticsClient(opts, { + networkMonitor: monitor, + }); + await client.init(); + + const debugInfo = await client.getDebugInfo(); + expect(debugInfo.networkStatus).toBe('connected'); + }); + + it('getDebugInfo reflects disconnected status', async () => { + const monitor = new StubNetworkMonitor('disconnected'); + const client = new MetaRouterAnalyticsClient(opts, { + networkMonitor: monitor, + }); + await client.init(); + + const debugInfo = await client.getDebugInfo(); + expect(debugInfo.networkStatus).toBe('disconnected'); + }); + + it('SDK functions normally when network monitoring unavailable', async () => { + // Don't inject networkMonitor — constructor will try to create + // a real NetworkMonitor which will fail in test env and fallback + // to always-connected + const client = new MetaRouterAnalyticsClient(opts); + await client.init(); + + client.track('Test Event'); + expect(client.queue).toHaveLength(1); + + await new Promise((resolve) => setTimeout(resolve, 10)); + await client.flush(); + expect(fetch).toHaveBeenCalled(); + }); + + it('offline -> online transition resets circuit breaker and triggers flush', async () => { + const monitor = new StubNetworkMonitor('connected'); + const client = new MetaRouterAnalyticsClient(opts, { + networkMonitor: monitor, + }); + await client.init(); + + // Go offline + monitor.simulate('disconnected'); + + // Track events while offline + client.track('Offline Event 1'); + client.track('Offline Event 2'); + + // Flush should not send (offline) + await client.flush(); + // Events should still be in queue since network is unavailable + // (the dispatcher guard prevents HTTP calls) + + // Go online — should trigger flush + (global as any).fetch = jest + .fn() + .mockResolvedValue({ ok: true, status: 200 }); + monitor.simulate('connected'); + + // Allow the async flush to complete + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(fetch).toHaveBeenCalled(); + }); + + it('reset() cleans up network monitor', async () => { + const monitor = new StubNetworkMonitor('connected'); + const stopSpy = jest.spyOn(monitor, 'stop'); + const client = new MetaRouterAnalyticsClient(opts, { + networkMonitor: monitor, + }); + await client.init(); + + await client.reset(); + expect(stopSpy).toHaveBeenCalled(); + }); + }); + describe('tracing', () => { it('does not include Trace header by default', async () => { const client = new MetaRouterAnalyticsClient(opts); diff --git a/src/analytics/MetaRouterAnalyticsClient.ts b/src/analytics/MetaRouterAnalyticsClient.ts index 63a68fc..066451d 100644 --- a/src/analytics/MetaRouterAnalyticsClient.ts +++ b/src/analytics/MetaRouterAnalyticsClient.ts @@ -13,6 +13,11 @@ import { import CircuitBreaker from './utils/circuitBreaker'; import Dispatcher from './dispatcher'; import { PersistentEventQueue } from './persistence/PersistentEventQueue'; +import { + NetworkMonitor, + type NetworkReachability, + type NetworkStatus, +} from './utils/networkMonitor'; /** * Analytics client for MetaRouter. @@ -37,12 +42,18 @@ export class MetaRouterAnalyticsClient { private dispatcher!: Dispatcher; private persistentQueue!: PersistentEventQueue; private tracingEnabled: boolean = false; + private networkMonitor: NetworkReachability; + private networkStatus: NetworkStatus = 'connected'; + private unsubscribeNetwork: (() => void) | null = null; /** * Initializes the analytics client with the provided options. * @param options - The initialization options. */ - constructor(options: InitOptions) { + constructor( + options: InitOptions, + deps?: { networkMonitor?: NetworkReachability } + ) { const { writeKey, ingestionHost, flushIntervalSeconds } = options; if (!writeKey || typeof writeKey !== 'string' || writeKey.trim() === '') { @@ -73,6 +84,7 @@ export class MetaRouterAnalyticsClient { setDebugLogging(options.debug ?? false); this.identityManager = new IdentityManager(); + this.networkMonitor = deps?.networkMonitor ?? new NetworkMonitor(); this.maxQueueBytes = options.maxQueueBytes ?? this.maxQueueBytes; this.dispatcher = new Dispatcher({ maxQueueBytes: this.maxQueueBytes, @@ -81,6 +93,7 @@ export class MetaRouterAnalyticsClient { flushIntervalSeconds: this.flushIntervalSeconds, baseRetryDelayMs: 1000, maxRetryDelayMs: 8000, + isNetworkAvailable: () => this.networkStatus === 'connected', endpoint: (path) => this.endpoint(path), fetchWithTimeout: (url, init, timeoutMs) => this.fetchWithTimeout(url, init, timeoutMs), @@ -159,6 +172,24 @@ export class MetaRouterAnalyticsClient { ); this.lifecycle = 'ready'; + + // Set initial network state and subscribe to changes + this.networkStatus = this.networkMonitor.currentStatus; + this.unsubscribeNetwork = this.networkMonitor.onStatusChange( + (status) => { + const wasOffline = this.networkStatus === 'disconnected'; + this.networkStatus = status; + + if (wasOffline && status === 'connected') { + log('Network connectivity restored — resuming flush'); + this.dispatcher.resetCircuitBreaker(); + void this.flush(); + } else if (status === 'disconnected') { + log('Network connectivity lost — pausing HTTP attempts'); + } + } + ); + log('MetaRouter SDK initialized'); // Flush immediately so rehydrated events ship on cold start @@ -459,6 +490,7 @@ export class MetaRouterAnalyticsClient { maxQueueBytes: d.maxQueueBytes, tracingEnabled: this.tracingEnabled, rehydratedEvents: this.persistentQueue.rehydratedEvents, + networkStatus: this.networkStatus, }; } @@ -479,6 +511,11 @@ export class MetaRouterAnalyticsClient { // Flip lifecycle first so other paths see we're resetting this.lifecycle = 'resetting'; + // Stop network monitoring + this.unsubscribeNetwork?.(); + this.unsubscribeNetwork = null; + this.networkMonitor.stop(); + // Stop background work this.dispatcher.stop(); this.appStateSubscription?.remove?.(); diff --git a/src/analytics/dispatcher.test.ts b/src/analytics/dispatcher.test.ts index 3111d24..94bf853 100644 --- a/src/analytics/dispatcher.test.ts +++ b/src/analytics/dispatcher.test.ts @@ -8,6 +8,7 @@ const baseOpts = () => ({ flushIntervalSeconds: 3600, // keep timer quiet unless started baseRetryDelayMs: 1000, maxRetryDelayMs: 8000, + isNetworkAvailable: () => true, endpoint: (p: string) => `https://example.com${p}`, fetchWithTimeout: jest.fn( async (_url?: string, _init?: any) => ({ ok: true, status: 200 }) as any @@ -326,6 +327,106 @@ describe('Dispatcher', () => { expect(d.getQueueSizeBytes()).toBe(0); }); + describe('network awareness', () => { + it('events enqueue while offline, no HTTP attempts', async () => { + const opts = baseOpts(); + opts.isNetworkAvailable = () => false; + const d = new Dispatcher(opts); + + d.enqueue({ type: 'track', event: 'e1' } as any); + d.enqueue({ type: 'track', event: 'e2' } as any); + await d.flush(); + + expect(opts.fetchWithTimeout).not.toHaveBeenCalled(); + expect(d.getQueueRef().length).toBe(2); + }); + + it('offline -> online triggers flush', async () => { + const opts = baseOpts(); + let online = false; + opts.isNetworkAvailable = () => online; + const d = new Dispatcher(opts); + + d.enqueue({ type: 'track', event: 'e1' } as any); + await d.flush(); + expect(opts.fetchWithTimeout).not.toHaveBeenCalled(); + + // Go online + online = true; + await d.flush(); + expect(opts.fetchWithTimeout).toHaveBeenCalled(); + expect(d.getQueueRef().length).toBe(0); + }); + + it('resetCircuitBreaker() resets circuit state', async () => { + const opts = baseOpts(); + opts.fetchWithTimeout = jest.fn( + async () => + ({ ok: false, status: 500, headers: { get: () => null } }) as any + ); + const d = new Dispatcher(opts); + + // Trip the circuit breaker + d.enqueue({ type: 'track', event: 'x' } as any); + await d.flush(); + d.enqueue({ type: 'track', event: 'y' } as any); + await d.flush(); + d.enqueue({ type: 'track', event: 'z' } as any); + await d.flush(); + + // Circuit should be impacted + expect(d.getDebugInfo().consecutiveRetries).toBeGreaterThan(0); + + // Reset + d.resetCircuitBreaker(); + expect(d.getDebugInfo().consecutiveRetries).toBe(0); + expect(d.getDebugInfo().circuitState).toBe('CLOSED'); + }); + + it('circuit breaker does NOT reset while still connected but failing', async () => { + const opts = baseOpts(); + opts.fetchWithTimeout = jest.fn( + async () => + ({ ok: false, status: 500, headers: { get: () => null } }) as any + ); + const d = new Dispatcher(opts); + + // Trip the circuit breaker (3 failures = threshold) + for (let i = 0; i < 3; i++) { + d.enqueue({ type: 'track', event: `e${i}` } as any); + await d.flush(); + } + + // Circuit should be OPEN (not auto-reset) + expect(d.getDebugInfo().circuitState).toBe('OPEN'); + }); + + it('getDebugInfo includes isNetworkAvailable', () => { + const opts = baseOpts(); + opts.isNetworkAvailable = () => false; + const d = new Dispatcher(opts); + expect(d.getDebugInfo().isNetworkAvailable).toBe(false); + }); + + it('offline log warns with queue count', async () => { + const opts = baseOpts(); + opts.isNetworkAvailable = () => false; + const d = new Dispatcher(opts); + + d.enqueue({ type: 'track', event: 'e1' } as any); + d.enqueue({ type: 'track', event: 'e2' } as any); + await d.flush(); + + expect( + (opts.warn as jest.Mock).mock.calls.some( + (c) => + String(c[0]).includes('Offline') && + String(c[0]).includes('2 event(s)') + ) + ).toBe(true); + }); + }); + it('stress test: all ~2K queued events successfully transmit in batches', async () => { const opts = baseOpts(); opts.autoFlushThreshold = 9999; // disable auto-flush diff --git a/src/analytics/dispatcher.ts b/src/analytics/dispatcher.ts index 39ae326..ee4ee67 100644 --- a/src/analytics/dispatcher.ts +++ b/src/analytics/dispatcher.ts @@ -11,6 +11,8 @@ export interface DispatcherOptions { baseRetryDelayMs: number; // retry floor base delay (default 1000) maxRetryDelayMs: number; // retry floor cap (default 8000) + isNetworkAvailable: () => boolean; // returns false when offline + endpoint: (path: string) => string; fetchWithTimeout: ( url: string, @@ -217,6 +219,13 @@ export default class Dispatcher { const doFlush = async () => { while (this.queue.length) { + if (!this.opts.isNetworkAvailable()) { + this.opts.warn( + `Offline — pausing HTTP attempts, ${this.queue.length} event(s) queued` + ); + return; + } + if (!this.circuit.allowRequest()) { const state = this.circuit.getState(); const wait = this.circuit.remainingCooldownMs(); @@ -380,6 +389,11 @@ export default class Dispatcher { return this.flushInFlight; } + resetCircuitBreaker(): void { + this.circuit.reset(); + this.consecutiveRetries = 0; + } + getDebugInfo() { return { queueLength: this.queue.length, @@ -391,6 +405,7 @@ export default class Dispatcher { maxQueueBytes: this.opts.maxQueueBytes, maxBatchSize: this.maxBatchSize, consecutiveRetries: this.consecutiveRetries, + isNetworkAvailable: this.opts.isNetworkAvailable(), }; } } diff --git a/src/analytics/persistence/__tests__/PersistentEventQueue.test.ts b/src/analytics/persistence/__tests__/PersistentEventQueue.test.ts index 3be4b56..6a05c57 100644 --- a/src/analytics/persistence/__tests__/PersistentEventQueue.test.ts +++ b/src/analytics/persistence/__tests__/PersistentEventQueue.test.ts @@ -23,6 +23,7 @@ function createDispatcher(overrides?: Partial) { autoFlushThreshold: 9999, maxBatchSize: 100, flushIntervalSeconds: 3600, + isNetworkAvailable: () => true, endpoint: (p: string) => `https://example.com${p}`, fetchWithTimeout: jest.fn(async () => ({ ok: true, status: 200 }) as any), canSend: () => true, diff --git a/src/analytics/types.ts b/src/analytics/types.ts index bfed08e..dd49392 100644 --- a/src/analytics/types.ts +++ b/src/analytics/types.ts @@ -33,6 +33,8 @@ export interface InitOptions { debug?: boolean; /** Max bytes (UTF-8) held in memory queue; oldest are dropped once cap is hit (default: 5MB) */ maxQueueBytes?: number; + /** Max events stored on disk during extended offline periods (default: 10000) */ + maxOfflineDiskEvents?: number; } export interface AnalyticsInterface { diff --git a/src/analytics/utils/circuitBreaker.test.ts b/src/analytics/utils/circuitBreaker.test.ts index ac95c8e..b6b374c 100644 --- a/src/analytics/utils/circuitBreaker.test.ts +++ b/src/analytics/utils/circuitBreaker.test.ts @@ -1,6 +1,6 @@ -import CircuitBreaker from "./circuitBreaker"; +import CircuitBreaker from './circuitBreaker'; -describe("CircuitBreaker", () => { +describe('CircuitBreaker', () => { let nowMs: number; let breaker: CircuitBreaker; let randomSpy: jest.SpyInstance | null; @@ -34,30 +34,30 @@ describe("CircuitBreaker", () => { } }); - it("starts CLOSED and allows requests", () => { + it('starts CLOSED and allows requests', () => { makeBreaker(); - expect(breaker.getState()).toBe("CLOSED"); + expect(breaker.getState()).toBe('CLOSED'); expect(breaker.allowRequest()).toBe(true); expect(breaker.nextAllowedAt()).toBe(nowMs); expect(breaker.remainingCooldownMs()).toBe(0); }); - it("trips OPEN after reaching failure threshold", () => { + it('trips OPEN after reaching failure threshold', () => { makeBreaker({ failureThreshold: 2, jitterRatio: 0 }); // First failure does not trip yet (threshold 2) breaker.onFailure(); - expect(breaker.getState()).toBe("CLOSED"); + expect(breaker.getState()).toBe('CLOSED'); // Second failure trips OPEN breaker.onFailure(); - expect(breaker.getState()).toBe("OPEN"); + expect(breaker.getState()).toBe('OPEN'); expect(breaker.allowRequest()).toBe(false); expect(breaker.nextAllowedAt()).toBe(nowMs + 1000); expect(breaker.remainingCooldownMs()).toBe(1000); }); - it("moves to HALF_OPEN after cooldown and allows limited probe(s)", () => { + it('moves to HALF_OPEN after cooldown and allows limited probe(s)', () => { makeBreaker({ failureThreshold: 1, jitterRatio: 0, @@ -66,11 +66,11 @@ describe("CircuitBreaker", () => { // Trip OPEN immediately breaker.onFailure(); - expect(breaker.getState()).toBe("OPEN"); + expect(breaker.getState()).toBe('OPEN'); // Cooldown elapses advanceTime(1000); - expect(breaker.getState()).toBe("HALF_OPEN"); // normalized + expect(breaker.getState()).toBe('HALF_OPEN'); // normalized // First probe allowed, second probe blocked until success/failure closes/reopens expect(breaker.allowRequest()).toBe(true); @@ -78,30 +78,30 @@ describe("CircuitBreaker", () => { // Success while HALF_OPEN closes and resets breaker.onSuccess(); - expect(breaker.getState()).toBe("CLOSED"); + expect(breaker.getState()).toBe('CLOSED'); expect(breaker.allowRequest()).toBe(true); }); - it("failed probe in HALF_OPEN reopens with exponential backoff", () => { + it('failed probe in HALF_OPEN reopens with exponential backoff', () => { makeBreaker({ failureThreshold: 1, cooldownMs: 1000, jitterRatio: 0 }); // First open breaker.onFailure(); - expect(breaker.getState()).toBe("OPEN"); + expect(breaker.getState()).toBe('OPEN'); expect(breaker.nextAllowedAt()).toBe(nowMs + 1000); // Move to HALF_OPEN advanceTime(1000); - expect(breaker.getState()).toBe("HALF_OPEN"); + expect(breaker.getState()).toBe('HALF_OPEN'); expect(breaker.allowRequest()).toBe(true); // Fail probe -> re-open with doubled backoff (2x) breaker.onFailure(); - expect(breaker.getState()).toBe("OPEN"); + expect(breaker.getState()).toBe('OPEN'); expect(breaker.nextAllowedAt()).toBe(nowMs + 2000); }); - it("exponential backoff is capped at maxCooldownMs", () => { + it('exponential backoff is capped at maxCooldownMs', () => { makeBreaker({ failureThreshold: 1, cooldownMs: 500, @@ -126,38 +126,38 @@ describe("CircuitBreaker", () => { expect(breaker.nextAllowedAt()).toBe(nowMs + 1000); }); - it("applies jitter within expected bounds", () => { + it('applies jitter within expected bounds', () => { // jitterRatio = 0.2, base = 1000 makeBreaker({ failureThreshold: 1, cooldownMs: 1000, jitterRatio: 0.2 }); // Force Math.random() = 1 -> +20% - randomSpy = jest.spyOn(Math, "random").mockReturnValue(1); + randomSpy = jest.spyOn(Math, 'random').mockReturnValue(1); breaker.onFailure(); - expect(breaker.getState()).toBe("OPEN"); + expect(breaker.getState()).toBe('OPEN'); // backoff = 1000, jitter = 200, random = 1 => backoff + 200 expect(breaker.nextAllowedAt()).toBe(nowMs + 1200); // Cooldown advanceTime(1200); - expect(breaker.getState()).toBe("HALF_OPEN"); + expect(breaker.getState()).toBe('HALF_OPEN'); expect(breaker.allowRequest()).toBe(true); // Fail probe with Math.random() = 0 -> -20%, base*2 = 2000 -> 1600 randomSpy.mockReturnValue(0); breaker.onFailure(); - expect(breaker.getState()).toBe("OPEN"); + expect(breaker.getState()).toBe('OPEN'); expect(breaker.nextAllowedAt()).toBe(nowMs + 1600); }); - it("getState normalizes OPEN to HALF_OPEN after cooldown without allowRequest()", () => { + it('getState normalizes OPEN to HALF_OPEN after cooldown without allowRequest()', () => { makeBreaker({ failureThreshold: 1, cooldownMs: 750, jitterRatio: 0 }); breaker.onFailure(); - expect(breaker.getState()).toBe("OPEN"); + expect(breaker.getState()).toBe('OPEN'); advanceTime(751); - expect(breaker.getState()).toBe("HALF_OPEN"); + expect(breaker.getState()).toBe('HALF_OPEN'); }); - it("respects halfOpenMaxConcurrent", () => { + it('respects halfOpenMaxConcurrent', () => { makeBreaker({ failureThreshold: 1, cooldownMs: 300, @@ -166,7 +166,7 @@ describe("CircuitBreaker", () => { }); breaker.onFailure(); advanceTime(300); - expect(breaker.getState()).toBe("HALF_OPEN"); + expect(breaker.getState()).toBe('HALF_OPEN'); // Two probes allowed, third blocked expect(breaker.allowRequest()).toBe(true); @@ -175,31 +175,31 @@ describe("CircuitBreaker", () => { // Close via success breaker.onSuccess(); - expect(breaker.getState()).toBe("CLOSED"); + expect(breaker.getState()).toBe('CLOSED'); }); - it("success resets consecutive failures and openCount after HALF_OPEN success", () => { + it('success resets consecutive failures and openCount after HALF_OPEN success', () => { makeBreaker({ failureThreshold: 2, cooldownMs: 500, jitterRatio: 0 }); // One failure below threshold breaker.onFailure(); - expect(breaker.getState()).toBe("CLOSED"); + expect(breaker.getState()).toBe('CLOSED'); // Success resets failure count breaker.onSuccess(); // Two fresh failures should trip OPEN again only after threshold breaker.onFailure(); - expect(breaker.getState()).toBe("CLOSED"); + expect(breaker.getState()).toBe('CLOSED'); breaker.onFailure(); - expect(breaker.getState()).toBe("OPEN"); + expect(breaker.getState()).toBe('OPEN'); // Move to HALF_OPEN and succeed -> close and reset openCount advanceTime(500); - expect(breaker.getState()).toBe("HALF_OPEN"); + expect(breaker.getState()).toBe('HALF_OPEN'); expect(breaker.allowRequest()).toBe(true); breaker.onSuccess(); - expect(breaker.getState()).toBe("CLOSED"); + expect(breaker.getState()).toBe('CLOSED'); // Failing again should use initial backoff (not compounded) // Need two failures to meet threshold (2) @@ -208,10 +208,67 @@ describe("CircuitBreaker", () => { expect(breaker.nextAllowedAt()).toBe(nowMs + 500); }); - it("treats failureThreshold < 1 as 1 (normalized)", () => { + it('treats failureThreshold < 1 as 1 (normalized)', () => { makeBreaker({ failureThreshold: 0, cooldownMs: 400, jitterRatio: 0 }); breaker.onFailure(); - expect(breaker.getState()).toBe("OPEN"); + expect(breaker.getState()).toBe('OPEN'); expect(breaker.nextAllowedAt()).toBe(nowMs + 400); }); + + describe('reset()', () => { + it('moves from OPEN to CLOSED', () => { + makeBreaker({ failureThreshold: 1, cooldownMs: 1000, jitterRatio: 0 }); + breaker.onFailure(); + expect(breaker.getState()).toBe('OPEN'); + + breaker.reset(); + expect(breaker.getState()).toBe('CLOSED'); + expect(breaker.allowRequest()).toBe(true); + expect(breaker.remainingCooldownMs()).toBe(0); + }); + + it('clears openCount (backoff escalation resets)', () => { + makeBreaker({ failureThreshold: 1, cooldownMs: 1000, jitterRatio: 0 }); + + // Trip multiple times to escalate backoff + breaker.onFailure(); // openCount=1, cooldown=1000 + advanceTime(1000); + breaker.allowRequest(); + breaker.onFailure(); // openCount=2, cooldown=2000 + advanceTime(2000); + breaker.allowRequest(); + breaker.onFailure(); // openCount=3, cooldown=4000 + + breaker.reset(); + + // Trip again — should use base cooldown (1000), not escalated + breaker.onFailure(); + expect(breaker.getState()).toBe('OPEN'); + expect(breaker.nextAllowedAt()).toBe(nowMs + 1000); + }); + + it('while CLOSED is a no-op', () => { + makeBreaker(); + expect(breaker.getState()).toBe('CLOSED'); + breaker.reset(); + expect(breaker.getState()).toBe('CLOSED'); + expect(breaker.allowRequest()).toBe(true); + }); + + it('clears consecutiveFailures', () => { + makeBreaker({ failureThreshold: 3, cooldownMs: 1000, jitterRatio: 0 }); + + // Accumulate 2 failures (below threshold of 3) + breaker.onFailure(); + breaker.onFailure(); + expect(breaker.getState()).toBe('CLOSED'); + + breaker.reset(); + + // Add 2 more failures — should still be CLOSED (not at threshold) + breaker.onFailure(); + breaker.onFailure(); + expect(breaker.getState()).toBe('CLOSED'); + }); + }); }); diff --git a/src/analytics/utils/circuitBreaker.ts b/src/analytics/utils/circuitBreaker.ts index bb31808..9d5b997 100644 --- a/src/analytics/utils/circuitBreaker.ts +++ b/src/analytics/utils/circuitBreaker.ts @@ -1,4 +1,4 @@ -export type CircuitState = "CLOSED" | "OPEN" | "HALF_OPEN"; +export type CircuitState = 'CLOSED' | 'OPEN' | 'HALF_OPEN'; export interface CircuitBreakerOptions { /** Failures in a row to trip OPEN */ @@ -28,10 +28,10 @@ export default class CircuitBreaker { private readonly jitterRatio: number; private readonly halfOpenMaxConcurrent: number; private readonly now: () => number; - private readonly onStateChange?: CircuitBreakerOptions["onStateChange"]; + private readonly onStateChange?: CircuitBreakerOptions['onStateChange']; private consecutiveFailures = 0; - private state: CircuitState = "CLOSED"; + private state: CircuitState = 'CLOSED'; private openUntil = 0; private halfOpenInFlight = 0; /** how many times we've opened (drives exponential backoff) */ @@ -55,17 +55,17 @@ export default class CircuitBreaker { const t = this.now(); // Normalize OPEN→HALF_OPEN when cooldown elapsed - if (this.state === "OPEN" && t >= this.openUntil) { + if (this.state === 'OPEN' && t >= this.openUntil) { this.halfOpenInFlight = 0; - this.setState("HALF_OPEN"); + this.setState('HALF_OPEN'); } - if (this.state === "OPEN") { + if (this.state === 'OPEN') { // Still cooling down return false; } - if (this.state === "HALF_OPEN") { + if (this.state === 'HALF_OPEN') { if (this.halfOpenInFlight >= this.halfOpenMaxConcurrent) return false; // Admit a probe this.halfOpenInFlight += 1; @@ -78,11 +78,11 @@ export default class CircuitBreaker { /** Call on successful request */ onSuccess(): void { - if (this.state === "HALF_OPEN") { + if (this.state === 'HALF_OPEN') { // One successful probe closes the circuit this.halfOpenInFlight = 0; this.openCount = 0; - this.setState("CLOSED"); + this.setState('CLOSED'); } // Success in CLOSED resets the consecutive failure counter this.consecutiveFailures = 0; @@ -92,14 +92,14 @@ export default class CircuitBreaker { onFailure(): void { this.consecutiveFailures += 1; - if (this.state === "HALF_OPEN") { + if (this.state === 'HALF_OPEN') { // failed probe → re-open with backoff this.tripOpen(); return; } if ( - this.state === "CLOSED" && + this.state === 'CLOSED' && this.consecutiveFailures >= this.failureThreshold ) { this.tripOpen(); @@ -115,14 +115,24 @@ export default class CircuitBreaker { /** Time when requests are next allowed (ms since epoch) */ nextAllowedAt(): number { const t = this.now(); - if (this.state === "OPEN") return this.openUntil; - if (this.state === "HALF_OPEN") { + if (this.state === 'OPEN') return this.openUntil; + if (this.state === 'HALF_OPEN') { // Capacity-limited in HALF_OPEN; return now (admission depends on allowRequest()) return t; } return t; // CLOSED } + /** Force-reset to CLOSED. Used on offline -> online transition + * to clear stale backoff from the offline period. */ + reset(): void { + this.setState('CLOSED'); + this.consecutiveFailures = 0; + this.openCount = 0; + this.openUntil = 0; + this.halfOpenInFlight = 0; + } + /** How long until next attempt (ms) */ remainingCooldownMs(): number { const t = this.now(); @@ -134,8 +144,8 @@ export default class CircuitBreaker { // Report HALF_OPEN if cooldown has elapsed, // but do not mutate state here — halfOpenInFlight // is reset when allowRequest() actually admits a probe. - if (this.state === "OPEN" && this.now() >= this.openUntil) { - return "HALF_OPEN"; + if (this.state === 'OPEN' && this.now() >= this.openUntil) { + return 'HALF_OPEN'; } return this.state; } @@ -175,6 +185,6 @@ export default class CircuitBreaker { const cooldown = Math.max(0, Math.floor(jittered)); this.openUntil = this.now() + cooldown; - this.setState("OPEN", { cooldownMs: cooldown }); + this.setState('OPEN', { cooldownMs: cooldown }); } } diff --git a/src/analytics/utils/networkMonitor.test.ts b/src/analytics/utils/networkMonitor.test.ts new file mode 100644 index 0000000..f6969eb --- /dev/null +++ b/src/analytics/utils/networkMonitor.test.ts @@ -0,0 +1,79 @@ +import { StubNetworkMonitor } from './networkMonitor'; + +describe('StubNetworkMonitor', () => { + it('defaults to connected', () => { + const monitor = new StubNetworkMonitor(); + expect(monitor.currentStatus).toBe('connected'); + }); + + it('accepts initial status', () => { + const monitor = new StubNetworkMonitor('disconnected'); + expect(monitor.currentStatus).toBe('disconnected'); + }); + + it('fires handler on status transition', () => { + const monitor = new StubNetworkMonitor(); + const handler = jest.fn(); + monitor.onStatusChange(handler); + + monitor.simulate('disconnected'); + expect(handler).toHaveBeenCalledWith('disconnected'); + expect(monitor.currentStatus).toBe('disconnected'); + }); + + it('does not fire handler when status unchanged', () => { + const monitor = new StubNetworkMonitor('connected'); + const handler = jest.fn(); + monitor.onStatusChange(handler); + + monitor.simulate('connected'); + expect(handler).not.toHaveBeenCalled(); + }); + + it('fires handler on each transition', () => { + const monitor = new StubNetworkMonitor(); + const handler = jest.fn(); + monitor.onStatusChange(handler); + + monitor.simulate('disconnected'); + monitor.simulate('connected'); + monitor.simulate('disconnected'); + + expect(handler).toHaveBeenCalledTimes(3); + expect(handler).toHaveBeenNthCalledWith(1, 'disconnected'); + expect(handler).toHaveBeenNthCalledWith(2, 'connected'); + expect(handler).toHaveBeenNthCalledWith(3, 'disconnected'); + }); + + it('stop() clears handler', () => { + const monitor = new StubNetworkMonitor(); + const handler = jest.fn(); + monitor.onStatusChange(handler); + + monitor.stop(); + monitor.simulate('disconnected'); + expect(handler).not.toHaveBeenCalled(); + }); + + it('unsubscribe clears handler', () => { + const monitor = new StubNetworkMonitor(); + const handler = jest.fn(); + const unsub = monitor.onStatusChange(handler); + + unsub(); + monitor.simulate('disconnected'); + expect(handler).not.toHaveBeenCalled(); + }); +}); + +describe('NetworkMonitor (graceful fallback)', () => { + it('defaults to connected when native module is unavailable', () => { + // NetworkMonitor constructor tries require('react-native').NativeModules + // In test environment, MetaRouterNetworkMonitor will not exist, + // so it should fall back to always-connected + const { NetworkMonitor } = require('./networkMonitor'); + const monitor = new NetworkMonitor(); + expect(monitor.currentStatus).toBe('connected'); + monitor.stop(); + }); +}); diff --git a/src/analytics/utils/networkMonitor.ts b/src/analytics/utils/networkMonitor.ts new file mode 100644 index 0000000..4f9d633 --- /dev/null +++ b/src/analytics/utils/networkMonitor.ts @@ -0,0 +1,119 @@ +import { warn } from './logger'; + +export type NetworkStatus = 'connected' | 'disconnected'; + +export interface NetworkReachability { + /** Current connectivity snapshot */ + readonly currentStatus: NetworkStatus; + /** Register callback for status transitions. Returns unsubscribe function. */ + onStatusChange(handler: (status: NetworkStatus) => void): () => void; + /** Tear down monitoring */ + stop(): void; +} + +export class NetworkMonitor implements NetworkReachability { + private _currentStatus: NetworkStatus = 'connected'; + private handler: ((status: NetworkStatus) => void) | null = null; + private unsubscribe: (() => void) | null = null; + private receivedNativeEvent = false; + + get currentStatus(): NetworkStatus { + return this._currentStatus; + } + + constructor() { + try { + const { NativeModules, NativeEventEmitter } = require('react-native'); + const MetaRouterNetworkMonitor = NativeModules.MetaRouterNetworkMonitor; + if (!MetaRouterNetworkMonitor) + throw new Error('Native module not available'); + + // Get initial state from native — skip if a live event already arrived + MetaRouterNetworkMonitor.getCurrentStatus() + .then((connected: boolean) => { + if (this.receivedNativeEvent) return; + const newStatus: NetworkStatus = connected + ? 'connected' + : 'disconnected'; + if (newStatus !== this._currentStatus) { + this._currentStatus = newStatus; + this.handler?.(newStatus); + } + }) + .catch((err: unknown) => { + warn( + 'Failed to get initial network status, defaulting to connected', + err + ); + }); + + // Subscribe to native connectivity events + const emitter = new NativeEventEmitter(MetaRouterNetworkMonitor); + const subscription = emitter.addListener( + 'onConnectivityChange', + (event: { isConnected: boolean }) => { + this.receivedNativeEvent = true; + const newStatus: NetworkStatus = event.isConnected + ? 'connected' + : 'disconnected'; + if (newStatus !== this._currentStatus) { + this._currentStatus = newStatus; + this.handler?.(newStatus); + } + } + ); + + this.unsubscribe = () => subscription.remove(); + } catch (err) { + warn( + 'Native network monitor not available, defaulting to connected', + err + ); + } + } + + onStatusChange(handler: (status: NetworkStatus) => void): () => void { + this.handler = handler; + return () => { + this.handler = null; + }; + } + + stop(): void { + this.unsubscribe?.(); + this.unsubscribe = null; + this.handler = null; + } +} + +export class StubNetworkMonitor implements NetworkReachability { + private _currentStatus: NetworkStatus; + private handler: ((status: NetworkStatus) => void) | null = null; + + get currentStatus(): NetworkStatus { + return this._currentStatus; + } + + constructor(initialStatus: NetworkStatus = 'connected') { + this._currentStatus = initialStatus; + } + + onStatusChange(handler: (status: NetworkStatus) => void): () => void { + this.handler = handler; + return () => { + this.handler = null; + }; + } + + stop(): void { + this.handler = null; + } + + /** Simulate a network transition from tests */ + simulate(status: NetworkStatus): void { + if (status !== this._currentStatus) { + this._currentStatus = status; + this.handler?.(status); + } + } +}