diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt index 614903b8f..be054b384 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt @@ -55,6 +55,7 @@ import com.onesignal.logger.IObservabilityEventRecorder import com.onesignal.logger.LoggerFactory import com.onesignal.notifications.INotificationsManager import com.onesignal.notifications.internal.MisconfiguredNotificationsManager +import com.onesignal.session.internal.session.operations.SessionStateModelStore import com.onesignal.user.internal.jwt.JwtTokenStore internal class CoreModule : IModule { @@ -92,6 +93,7 @@ internal class CoreModule : IModule { // Operations builder.register().provides() + builder.register().provides() builder.register() .provides() .provides() diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationModelStore.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationModelStore.kt index c3d6e4591..6235e6e32 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationModelStore.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationModelStore.kt @@ -4,6 +4,9 @@ import com.onesignal.common.modeling.ModelStore import com.onesignal.core.internal.operations.Operation import com.onesignal.core.internal.preferences.IPreferencesService import com.onesignal.debug.internal.logging.Logging +import com.onesignal.session.internal.session.operations.CreateSessionOperation +import com.onesignal.session.internal.session.operations.UpdateSessionOperation +import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor import com.onesignal.user.internal.operations.CreateSubscriptionOperation import com.onesignal.user.internal.operations.DeleteAliasOperation import com.onesignal.user.internal.operations.DeleteSubscriptionOperation @@ -63,6 +66,8 @@ internal class OperationModelStore(prefs: IPreferencesService) : ModelStore TrackSessionEndOperation() UpdateUserOperationExecutor.TRACK_PURCHASE -> TrackPurchaseOperation() CustomEventOperationExecutor.CUSTOM_EVENT -> TrackCustomEventOperation() + SessionOperationExecutor.CREATE_SESSION -> CreateSessionOperation() + SessionOperationExecutor.UPDATE_SESSION -> UpdateSessionOperation() else -> throw Exception("Unrecognized operation: $operationName") } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt index 46b30a452..4502ff5c8 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/OperationRepo.kt @@ -16,6 +16,8 @@ import com.onesignal.core.internal.startup.IStartableService import com.onesignal.core.internal.time.ITime import com.onesignal.debug.LogLevel import com.onesignal.debug.internal.logging.Logging +import com.onesignal.session.internal.session.operations.CreateSessionOperation +import com.onesignal.session.internal.session.operations.SessionStateModelStore import com.onesignal.user.internal.jwt.JwtRequirement import com.onesignal.user.internal.jwt.JwtTokenStore import com.onesignal.user.internal.operations.LoginUserOperation @@ -33,6 +35,7 @@ import kotlin.math.min import kotlin.random.Random import kotlin.reflect.KClass +@Suppress("LongParameterList") internal class OperationRepo( executors: List, private val _operationModelStore: OperationModelStore, @@ -41,6 +44,7 @@ internal class OperationRepo( private val _newRecordState: NewRecordsState, private val _jwtTokenStore: JwtTokenStore, private val _identityVerificationService: IdentityVerificationService, + sessionStateStore: SessionStateModelStore, ) : IOperationRepo, IStartableService { internal class OperationQueueItem( @@ -107,6 +111,11 @@ internal class OperationRepo( private val inFlightLogins = mutableMapOf() private val waiter = WaiterWithValue() private var paused = false + private val sessionQueuePolicy = + SessionQueuePolicy( + awaitingConsent = { _configModelStore.model.let { it.consentRequired == true && it.consentGiven != true } }, + sessions = sessionStateStore, + ) // Guarded by queue. Set by a force that lands while ops are out of the queue executing. private var forcedDuringExecute = false @@ -187,7 +196,7 @@ internal class OperationRepo( operation.id = UUID.randomUUID().toString() scope.launch { - internalEnqueue(OperationQueueItem(operation, bucket = enqueueIntoBucket), flush, true) + enqueueNew(OperationQueueItem(operation, bucket = enqueueIntoBucket), flush) } } @@ -202,7 +211,7 @@ internal class OperationRepo( operation.id = UUID.randomUUID().toString() val waiter = WaiterWithValue() scope.launch { - internalEnqueue(OperationQueueItem(operation, waiter, bucket = enqueueIntoBucket), flush, true) + enqueueNew(OperationQueueItem(operation, waiter, bucket = enqueueIntoBucket), flush) } val timeoutMs = _configModelStore.model.opRepoAwaitTimeout return withTimeoutOrNull(timeoutMs) { @@ -218,12 +227,14 @@ internal class OperationRepo( * they cannot be authenticated and would otherwise sit in the queue forever, blocked by * `hasValidJwtIfRequired`. LoginUserOperation is exempt because it's enqueued * intentionally during logout and purged later by [removeOperationsWithoutExternalId] - * if needed. Outer-gated on `newCodePathsRun` so Phase 1 customers stay byte-for-byte + * if needed. Ops with `requiresJwt = false` are exempt since they can still execute, except a + * session create that waits on an anonymous user or subscription that will never be created. + * Outer-gated on `newCodePathsRun` so Phase 1 customers stay byte-for-byte * on the legacy enqueue path. */ private fun shouldSuppressAnonymousOp(op: Operation): Boolean { if (!_identityVerificationService.newCodePathsRun) return false - if (op is LoginUserOperation) return false + if (op is LoginUserOperation || !(op.requiresJwt || op.waitsForAnonymousUser)) return false val suppress = _configModelStore.model.useIdentityVerification == JwtRequirement.REQUIRED && op.externalId == null @@ -233,6 +244,23 @@ internal class OperationRepo( return suppress } + private fun enqueueNew( + queueItem: OperationQueueItem, + flush: Boolean, + ) { + val accepted = + synchronized(queue) { + val result = sessionQueuePolicy.onEnqueue(queue, queueItem) + dropAndWake(result.removed) + result.accept + } + if (accepted) { + internalEnqueue(queueItem, flush, true) + } else { + queueItem.wakeWaiters(OperationWaitResult(false)) + } + } + /** * Only used inside this class, adds OperationQueueItem to queue * WARNING: Never set flush=true until budget rules are added, even for internal use! @@ -353,16 +381,18 @@ internal class OperationRepo( } /** - * Drops queued operations whose externalId is null. Called by the IV-aware HYDRATE + * Drops queued operations whose externalId is null and that require a JWT, or that wait for an + * anonymous user, which is never created under IV. Called by the IV-aware HYDRATE * choreography in [OperationRepoIvExtensions] when `jwt_required` becomes REQUIRED * to evict anon ops that can no longer execute. */ internal fun removeOperationsWithoutExternalId() { val removedIds: List = synchronized(queue) { - val anonymous = queue.filter { it.operation.externalId == null } + val unsendable = queue.filter { it.operation.externalId == null && (it.operation.requiresJwt || it.operation.waitsForAnonymousUser) } + queue.removeAll(unsendable) + val anonymous = unsendable + unsendable.flatMap { sessionQueuePolicy.onDropped(queue, it.operation) } anonymous.forEach { it.wakeWaiters(OperationWaitResult(false)) } - queue.removeAll(anonymous) // IV=ON never transfers anonymous state; clear existingOnesignalId so the // executor takes the createUser (upsert) path. The merge-anon-into-identified // path can't dispatch under IV — anon user creation requires a JWT-less call @@ -382,6 +412,10 @@ internal class OperationRepo( removedIds.forEach { _operationModelStore.remove(it) } } + // Session ops take no JWT, but a create can't be sent until its user and subscription exist. + private val Operation.waitsForAnonymousUser: Boolean + get() = this is CreateSessionOperation && (IDManager.isLocalId(onesignalId) || IDManager.isLocalId(subscriptionId)) + /** * Post-HYDRATE maintenance: scheduled on IO so it runs *after* `loadSavedOperations` * populates the queue (fix for an earlier race where the purge ran against an empty @@ -454,6 +488,7 @@ internal class OperationRepo( ops.forEach { it.operation.translateIds(response.idTranslations) } synchronized(queue) { queue.forEach { it.operation.translateIds(response.idTranslations) } + sessionQueuePolicy.recordTranslations(response.idTranslations) } response.idTranslations.values.forEach { _newRecordState.add(it) } } @@ -560,7 +595,14 @@ internal class OperationRepo( } if (response.idTranslations != null) { - delayForPostCreate(_configModelStore.model.opRepoPostCreateDelay) + val postCreateDelay = _configModelStore.model.opRepoPostCreateDelay + // A new session's only dependents are its updates, which NewRecordsState holds back, + // so only the wake is delayed. + if (ops.all { it.operation is CreateSessionOperation }) { + scope.launch { delayForPostCreate(postCreateDelay) } + } else { + delayForPostCreate(postCreateDelay) + } } } catch (e: Throwable) { Logging.log(LogLevel.ERROR, "Error attempting to execute operation: $ops", e) @@ -582,6 +624,8 @@ internal class OperationRepo( ) { ops.forEach { _operationModelStore.remove(it.operation.id) } ops.forEach { it.wakeWaiters(response.toWaitResult(false)) } + val orphans = synchronized(queue) { ops.flatMap { sessionQueuePolicy.onDropped(queue, it.operation) } } + if (orphans.isNotEmpty()) dropAndWake(orphans) } /** diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/SessionQueuePolicy.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/SessionQueuePolicy.kt new file mode 100644 index 000000000..8205f23b3 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/operations/impl/SessionQueuePolicy.kt @@ -0,0 +1,187 @@ +package com.onesignal.core.internal.operations.impl + +import com.onesignal.common.IDManager +import com.onesignal.common.modeling.SimpleModelStore +import com.onesignal.core.internal.operations.Operation +import com.onesignal.debug.internal.logging.Logging +import com.onesignal.session.internal.session.operations.CreateSessionOperation +import com.onesignal.session.internal.session.operations.SessionStateModel +import com.onesignal.session.internal.session.operations.UpdateSessionOperation + +/** + * Queue rules for sessions API operations in [OperationRepo]. Every function must be called while + * holding the queue lock. Functions return the queue items they removed; the caller drops them + * from the store and wakes their waiters. + */ +internal class SessionQueuePolicy( + private val maxQueued: Int = MAX_QUEUED_SESSION_OPERATIONS, + // Without consent HttpClient fails every request, so these must not be stored or sent. + private val awaitingConsent: () -> Boolean = { false }, + private val sessions: SimpleModelStore = SimpleModelStore({ SessionStateModel() }), +) { + // Recent local -> backend ID translations, for an operation built from a local ID just before + // its record was created but enqueued after the queue was translated. + private val recentTranslations = + object : LinkedHashMap() { + override fun removeEldestEntry(eldest: MutableMap.MutableEntry?) = size > MAX_RECENT_TRANSLATIONS + } + + fun recordTranslations(idTranslations: Map) { + recentTranslations.putAll(idTranslations) + sessions.list().forEach { it.translateIds(idTranslations) } + } + + class EnqueueResult( + val accept: Boolean, + val removed: List = emptyList(), + ) + + fun onEnqueue( + queue: MutableList, + incoming: OperationRepo.OperationQueueItem, + ): EnqueueResult { + val op = incoming.operation + if (!op.isSessionOperation) return EnqueueResult(accept = true) + op.translateIds(recentTranslations) + + val result = + when { + awaitingConsent() -> { + Logging.debug("SessionQueuePolicy: privacy consent not given, dropping $op") + EnqueueResult(accept = false) + } + op is UpdateSessionOperation -> admitUpdate(queue, incoming, op) + else -> admit(queue, incoming) + } + // Its updates could never be sent. + if (!result.accept && op is CreateSessionOperation) updateState(op.localSessionId, queue) { it.isClosed = true } + return result + } + + private fun admitUpdate( + queue: MutableList, + incoming: OperationRepo.OperationQueueItem, + op: UpdateSessionOperation, + ): EnqueueResult { + val state = stateFor(op.sessionId) + state?.serverSessionId?.let { op.translateIds(mapOf(op.sessionId to it)) } + val rejection = + when { + state?.isClosed == true || hasQueuedEnd(queue, op.sessionId) -> "is closed" + IDManager.isLocalId(op.sessionId) && state?.onesignalId == null -> "has no create" + else -> null + } + if (rejection != null) { + Logging.debug("SessionQueuePolicy: session ${op.sessionId} $rejection, dropping $op") + return EnqueueResult(accept = false) + } + state?.let { op.useCreateIds(it.onesignalId, it.subscriptionId) } + return admit(queue, incoming) + } + + private fun admit( + queue: MutableList, + incoming: OperationRepo.OperationQueueItem, + ): EnqueueResult { + val op = incoming.operation + val removed = mutableListOf() + if (op is UpdateSessionOperation) { + val unsent = queue.filter { (it.operation as? UpdateSessionOperation)?.sessionId == op.sessionId } + // Durations are cumulative, so the newest update carries everything the unsent ones would. + unsent.forEach { + val replaced = it.operation as UpdateSessionOperation + op.activeDuration = maxOf(op.activeDuration, replaced.activeDuration) + incoming.retries = maxOf(incoming.retries, it.retries) + incoming.retryAt = maxOf(incoming.retryAt, it.retryAt) + val waiters = synchronized(it) { it.waiters.toList().also { _ -> it.waiters.clear() } } + incoming.attachOrWake(waiters) + } + queue.removeAll(unsent) + removed.addAll(unsent) + } + + var accept = true + while (queue.count { it.operation.isSessionOperation } + (if (accept) 1 else 0) > maxQueued) { + val oldestSession = queue.first { it.operation.isSessionOperation }.operation.sessionKey + val dropped = queue.filter { it.operation.sessionKey == oldestSession } + Logging.warn("SessionQueuePolicy: more than $maxQueued sessions API operations queued, dropping session $oldestSession") + queue.removeAll(dropped) + removed.addAll(dropped) + if (op.sessionKey == oldestSession) accept = false + } + if (accept && op is CreateSessionOperation) { + updateState(op.localSessionId, queue) { + it.onesignalId = op.onesignalId + it.subscriptionId = op.subscriptionId + } + } + if (accept && op is UpdateSessionOperation && op.isEnd) updateState(op.sessionId, queue) { it.isClosed = true } + return EnqueueResult(accept, removed) + } + + /** Removes and rejects updates that can never send because [dropped] was their session's create. */ + fun onDropped( + queue: MutableList, + dropped: Operation, + ): List { + if (dropped !is CreateSessionOperation) return emptyList() + updateState(dropped.localSessionId, queue) { it.isClosed = true } + val orphans = queue.filter { (it.operation as? UpdateSessionOperation)?.sessionId == dropped.localSessionId } + if (orphans.isNotEmpty()) { + Logging.debug("SessionQueuePolicy: create for session ${dropped.localSessionId} was dropped, dropping ${orphans.size} updates") + queue.removeAll(orphans) + } + return orphans + } + + private fun stateFor(sessionId: String) = sessions.list().firstOrNull { it.id == sessionId || it.serverSessionId == sessionId } + + private fun updateState( + sessionKey: String, + queue: List, + change: (SessionStateModel) -> Unit, + ) { + val existing = stateFor(sessionKey) + if (existing != null) { + change(existing) + } else { + sessions.add(SessionStateModel().also { it.id = sessionKey }.also(change)) + } + trimSessions(queue, sessionKey) + } + + /** Drops the oldest sessions past the cap, but never one with operations still queued, which need its state. */ + private fun trimSessions( + queue: List, + keep: String, + ) { + val all = sessions.list() + val excess = all.size - maxQueued + if (excess <= 0) return + val queuedSessions = queue.mapNotNull { it.operation.sessionKey }.toSet() + keep + all + .filter { it.id !in queuedSessions && it.serverSessionId !in queuedSessions } + .take(excess) + .forEach { sessions.remove(it.id) } + } + + private fun hasQueuedEnd( + queue: List, + sessionId: String, + ) = queue.any { (it.operation as? UpdateSessionOperation)?.let { update -> update.sessionId == sessionId && update.isEnd } == true } + + private val Operation.isSessionOperation get() = this is CreateSessionOperation || this is UpdateSessionOperation + + private val Operation.sessionKey: String? + get() = + when (this) { + is CreateSessionOperation -> localSessionId + is UpdateSessionOperation -> sessionId + else -> null + } + + companion object { + const val MAX_QUEUED_SESSION_OPERATIONS = 100 + private const val MAX_RECENT_TRANSLATIONS = 50 + } +} diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/SessionModule.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/SessionModule.kt index 1dc8cc988..dcee3dc23 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/SessionModule.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/SessionModule.kt @@ -3,6 +3,7 @@ package com.onesignal.session import com.onesignal.common.modules.IModule import com.onesignal.common.services.ServiceBuilder import com.onesignal.core.internal.background.IBackgroundService +import com.onesignal.core.internal.operations.IOperationExecutor import com.onesignal.core.internal.startup.IBootstrapService import com.onesignal.core.internal.startup.IStartableService import com.onesignal.session.internal.SessionManager @@ -22,6 +23,7 @@ import com.onesignal.session.internal.session.backend.ISessionsBackendService import com.onesignal.session.internal.session.backend.impl.SessionsBackendService import com.onesignal.session.internal.session.impl.SessionListener import com.onesignal.session.internal.session.impl.SessionService +import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor internal class SessionModule : IModule { override fun register(builder: ServiceBuilder) { @@ -46,5 +48,6 @@ internal class SessionModule : IModule { builder.register().provides() builder.register().provides() builder.register().provides() + builder.register().provides() } } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/SessionModel.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/SessionModel.kt index 9f520e059..dc55546b3 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/SessionModel.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/SessionModel.kt @@ -97,11 +97,11 @@ class SessionModel : Model() { } /** - * Null until the sessions API creates this session. Cleared when a new session starts. + * Identifies this session's sessions API requests. A new local ID for each session. */ - var serverSessionId: String? - get() = getOptStringProperty(::serverSessionId.name) + var localSessionId: String + get() = getStringProperty(::localSessionId.name) { "" } set(value) { - setOptStringProperty(::serverSessionId.name, value) + setStringProperty(::localSessionId.name, value) } } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/impl/SessionService.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/impl/SessionService.kt index 7ac57d91b..a2a45bdfb 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/impl/SessionService.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/impl/SessionService.kt @@ -197,7 +197,7 @@ internal class SessionService( .takeIf { it.hasProperty(IdentityConstants.ONESIGNAL_ID) } ?.onesignalId session.subscriptionId = config?.pushSubscriptionId - session.serverSessionId = null + session.localSessionId = IDManager.createLocalId() session.isValid = true Logging.debug("SessionService: New session started at ${session.startTime}") sessionLifeCycleNotifier.fire { it.onSessionStarted() } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/CreateSessionOperation.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/CreateSessionOperation.kt new file mode 100644 index 000000000..c4daa9c3d --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/CreateSessionOperation.kt @@ -0,0 +1,85 @@ +package com.onesignal.session.internal.session.operations + +import com.onesignal.common.IDManager +import com.onesignal.core.internal.operations.GroupComparisonType +import com.onesignal.core.internal.operations.Operation +import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor + +/** + * An [Operation] to create a session through the sessions API. + */ +internal class CreateSessionOperation() : Operation(SessionOperationExecutor.CREATE_SESSION) { + var appId: String + get() = getStringProperty(::appId.name) + private set(value) { + setStringProperty(::appId.name, value) + } + + /** + * Stands in for the backend session ID until the create succeeds. Updates queued before then + * reference it and are translated to the backend ID. + */ + var localSessionId: String + get() = getStringProperty(::localSessionId.name) + private set(value) { + setStringProperty(::localSessionId.name, value) + } + + var onesignalId: String + get() = getStringProperty(::onesignalId.name) + private set(value) { + setStringProperty(::onesignalId.name, value) + } + + var subscriptionId: String + get() = getStringProperty(::subscriptionId.name) + private set(value) { + setStringProperty(::subscriptionId.name, value) + } + + /** Unix time in milliseconds. */ + var startTime: Long + get() = getLongProperty(::startTime.name) + private set(value) { + setLongProperty(::startTime.name, value) + } + + /** Stable across retries and restarts, so the backend recognizes a resent create. */ + val idempotencyKey: String get() = localSessionId + + /** Responses so far that had no session_id. */ + var missingSessionIdAttempts: Int + get() = getOptIntProperty(::missingSessionIdAttempts.name) ?: 0 + internal set(value) { + setIntProperty(::missingSessionIdAttempts.name, value) + } + + override val requiresJwt: Boolean get() = false + override val createComparisonKey: String get() = "" + override val modifyComparisonKey: String get() = "" + override val groupComparisonType: GroupComparisonType = GroupComparisonType.NONE + override val canStartExecute: Boolean + get() = !IDManager.isLocalId(onesignalId) && !IDManager.isLocalId(subscriptionId) + override val applyToRecordId: String get() = localSessionId + + constructor( + appId: String, + localSessionId: String, + onesignalId: String, + subscriptionId: String, + startTime: Long, + externalId: String?, + ) : this() { + this.appId = appId + this.localSessionId = localSessionId + this.onesignalId = onesignalId + this.subscriptionId = subscriptionId + this.startTime = startTime + this.externalId = externalId + } + + override fun translateIds(map: Map) { + map[onesignalId]?.let { onesignalId = it } + map[subscriptionId]?.let { subscriptionId = it } + } +} diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/SessionStateModel.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/SessionStateModel.kt new file mode 100644 index 000000000..93fbd1b8f --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/SessionStateModel.kt @@ -0,0 +1,49 @@ +package com.onesignal.session.internal.session.operations + +import com.onesignal.common.modeling.Model +import com.onesignal.common.modeling.SimpleModelStore +import com.onesignal.core.internal.preferences.IPreferencesService + +/** + * The one place that knows whether a session was created and whether it ended. Persisted, since a + * session's operations can leave the queue, or the process can restart, while more of its updates + * are still coming. [id] is the session's local ID, or its backend ID when only that is known. + */ +internal class SessionStateModel : Model() { + /** Set once the create succeeds. */ + var serverSessionId: String? + get() = getOptStringProperty(::serverSessionId.name) + set(value) { + setOptStringProperty(::serverSessionId.name, value) + } + + /** The user the create is sent with, which every update repeats. Null when no create is known. */ + var onesignalId: String? + get() = getOptStringProperty(::onesignalId.name) + set(value) { + setOptStringProperty(::onesignalId.name, value) + } + + /** The subscription the create is sent with, which every update repeats. */ + var subscriptionId: String? + get() = getOptStringProperty(::subscriptionId.name) + set(value) { + setOptStringProperty(::subscriptionId.name, value) + } + + /** Takes no more updates: it ended, or its create was dropped. */ + var isClosed: Boolean + get() = getBooleanProperty(::isClosed.name) { false } + set(value) { + setBooleanProperty(::isClosed.name, value) + } + + fun translateIds(map: Map) { + if (serverSessionId == null) map[id]?.let { serverSessionId = it } + onesignalId?.let { map[it] }?.let { onesignalId = it } + subscriptionId?.let { map[it] }?.let { subscriptionId = it } + } +} + +internal class SessionStateModelStore(prefs: IPreferencesService) : + SimpleModelStore({ SessionStateModel() }, "sessionsApiState", prefs) diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt new file mode 100644 index 000000000..534005708 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/UpdateSessionOperation.kt @@ -0,0 +1,104 @@ +package com.onesignal.session.internal.session.operations + +import com.onesignal.common.IDManager +import com.onesignal.core.internal.operations.GroupComparisonType +import com.onesignal.core.internal.operations.Operation +import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor +import java.util.UUID + +/** + * An [Operation] to report a session's cumulative duration through the sessions API, and to end + * the session when [endTime] is set. + */ +internal class UpdateSessionOperation() : Operation(SessionOperationExecutor.UPDATE_SESSION) { + var appId: String + get() = getStringProperty(::appId.name) + private set(value) { + setStringProperty(::appId.name, value) + } + + /** + * The backend session ID, or the [CreateSessionOperation.localSessionId] while that create is + * still queued. + */ + var sessionId: String + get() = getStringProperty(::sessionId.name) + private set(value) { + setStringProperty(::sessionId.name, value) + } + + var onesignalId: String + get() = getStringProperty(::onesignalId.name) + private set(value) { + setStringProperty(::onesignalId.name, value) + } + + var subscriptionId: String + get() = getStringProperty(::subscriptionId.name) + private set(value) { + setStringProperty(::subscriptionId.name, value) + } + + /** Cumulative active time of the session, in milliseconds. */ + var activeDuration: Long + get() = getLongProperty(::activeDuration.name) + internal set(value) { + setLongProperty(::activeDuration.name, value) + } + + /** Unix time in milliseconds. Null for a duration-only update. */ + var endTime: Long? + get() = getOptLongProperty(::endTime.name) + private set(value) { + setOptLongProperty(::endTime.name, value) + } + + /** Persisted so a retry after a restart is recognized by the backend as the same request. */ + var idempotencyKey: String + get() = getStringProperty(::idempotencyKey.name) + private set(value) { + setStringProperty(::idempotencyKey.name, value) + } + + val isEnd: Boolean get() = endTime != null + + override val requiresJwt: Boolean get() = false + override val createComparisonKey: String get() = "" + override val modifyComparisonKey: String get() = "" + override val groupComparisonType: GroupComparisonType = GroupComparisonType.NONE + override val canStartExecute: Boolean + get() = !IDManager.isLocalId(sessionId) && !IDManager.isLocalId(onesignalId) && !IDManager.isLocalId(subscriptionId) + override val applyToRecordId: String get() = sessionId + + constructor( + appId: String, + sessionId: String, + onesignalId: String, + subscriptionId: String, + activeDuration: Long, + endTime: Long? = null, + ) : this() { + this.appId = appId + this.sessionId = sessionId + this.onesignalId = onesignalId + this.subscriptionId = subscriptionId + this.activeDuration = activeDuration + this.endTime = endTime + this.idempotencyKey = UUID.randomUUID().toString() + } + + /** The backend requires an update to repeat the user and subscription its create sent. */ + fun useCreateIds( + onesignalId: String?, + subscriptionId: String?, + ) { + onesignalId?.let { this.onesignalId = it } + subscriptionId?.let { this.subscriptionId = it } + } + + override fun translateIds(map: Map) { + map[sessionId]?.let { sessionId = it } + map[onesignalId]?.let { onesignalId = it } + map[subscriptionId]?.let { subscriptionId = it } + } +} diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt new file mode 100644 index 000000000..bf716f63a --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/operations/impl/SessionOperationExecutor.kt @@ -0,0 +1,121 @@ +package com.onesignal.session.internal.session.operations.impl + +import com.onesignal.common.exceptions.BackendException +import com.onesignal.core.internal.device.IDeviceService +import com.onesignal.core.internal.operations.ExecutionResponse +import com.onesignal.core.internal.operations.ExecutionResult +import com.onesignal.core.internal.operations.IOperationExecutor +import com.onesignal.core.internal.operations.Operation +import com.onesignal.debug.internal.logging.Logging +import com.onesignal.session.internal.session.backend.CreateSessionRequest +import com.onesignal.session.internal.session.backend.ISessionsBackendService +import com.onesignal.session.internal.session.backend.UpdateSessionRequest +import com.onesignal.session.internal.session.operations.CreateSessionOperation +import com.onesignal.session.internal.session.operations.UpdateSessionOperation +import com.onesignal.user.internal.operations.impl.states.NewRecordsState +import java.util.concurrent.TimeUnit + +internal class SessionOperationExecutor( + private val sessionsBackend: ISessionsBackendService, + private val deviceService: IDeviceService, + private val newRecordState: NewRecordsState, +) : IOperationExecutor { + override val operations: List + get() = listOf(CREATE_SESSION, UPDATE_SESSION) + + override suspend fun execute(operations: List): ExecutionResponse = + when (val operation = operations.single()) { + is CreateSessionOperation -> createSession(operation) + is UpdateSessionOperation -> updateSession(operation) + else -> error("Unrecognized operation: $operation") + } + + private suspend fun createSession(operation: CreateSessionOperation): ExecutionResponse { + val sessionId = + try { + sessionsBackend.createSession( + operation.appId, + CreateSessionRequest( + onesignalId = operation.onesignalId, + subscriptionId = operation.subscriptionId, + deviceType = deviceService.deviceType.value, + startTime = TimeUnit.MILLISECONDS.toSeconds(operation.startTime), + idempotencyKey = operation.idempotencyKey, + ), + ) + } catch (ex: BackendException) { + val retryMissing = + newRecordState.isInMissingRetryWindow(operation.onesignalId) || + newRecordState.isInMissingRetryWindow(operation.subscriptionId) + return failure(ex, retryMissing) + } + + return if (sessionId == null) { + missingSessionId(operation) + } else { + ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(operation.localSessionId to sessionId)) + } + } + + /** A retry with the same idempotency key should return the session the backend already created. */ + private fun missingSessionId(operation: CreateSessionOperation): ExecutionResponse { + operation.missingSessionIdAttempts++ + return if (operation.missingSessionIdAttempts < MAX_MISSING_SESSION_ID_ATTEMPTS) { + Logging.warn("SessionOperationExecutor: create session response has no session_id, retrying $operation") + ExecutionResponse(ExecutionResult.FAIL_RETRY) + } else { + Logging.warn("SessionOperationExecutor: create session responses never had a session_id, dropping $operation") + ExecutionResponse(ExecutionResult.FAIL_NORETRY) + } + } + + private suspend fun updateSession(operation: UpdateSessionOperation): ExecutionResponse { + try { + sessionsBackend.updateSession( + operation.appId, + operation.sessionId, + UpdateSessionRequest( + onesignalId = operation.onesignalId, + subscriptionId = operation.subscriptionId, + durationSeconds = TimeUnit.MILLISECONDS.toSeconds(operation.activeDuration), + idempotencyKey = operation.idempotencyKey, + endTime = operation.endTime?.let { TimeUnit.MILLISECONDS.toSeconds(it) }, + ), + ) + } catch (ex: BackendException) { + // The create's 202 is processed asynchronously, so a fresh session can briefly 404. + return failure(ex, retryMissing = newRecordState.isInMissingRetryWindow(operation.sessionId)) + } + return ExecutionResponse(ExecutionResult.SUCCESS) + } + + private fun failure( + ex: BackendException, + retryMissing: Boolean, + ): ExecutionResponse { + val code = ex.statusCode + val retryable = + // HttpClient reports offline and other network errors as 0 or -1. + code <= 0 || + code == HTTP_REQUEST_TIMEOUT || + code == HTTP_TOO_MANY_REQUESTS || + code >= HTTP_SERVER_ERROR || + (code == HTTP_NOT_FOUND && retryMissing) + return if (retryable) { + ExecutionResponse(ExecutionResult.FAIL_RETRY, retryAfterSeconds = ex.retryAfterSeconds) + } else { + ExecutionResponse(ExecutionResult.FAIL_NORETRY, httpStatusCode = code, httpResponse = ex.response) + } + } + + companion object { + const val CREATE_SESSION = "create-session" + const val UPDATE_SESSION = "update-session" + const val MAX_MISSING_SESSION_ID_ATTEMPTS = 3 + + private const val HTTP_NOT_FOUND = 404 + private const val HTTP_REQUEST_TIMEOUT = 408 + private const val HTTP_TOO_MANY_REQUESTS = 429 + private const val HTTP_SERVER_ERROR = 500 + } +} diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/application/SDKInitTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/application/SDKInitTests.kt index 7f5c9e855..daa182c08 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/application/SDKInitTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/application/SDKInitTests.kt @@ -83,11 +83,7 @@ class SDKInitTests : FunSpec({ error.get() shouldBe null } - beforeAny { - Logging.logLevel = LogLevel.NONE - } - - afterAny { + fun clearPrefs() { val context = getApplicationContext() val prefs = context.getSharedPreferences("OneSignal", Context.MODE_PRIVATE) prefs.edit() @@ -101,6 +97,16 @@ class SDKInitTests : FunSpec({ .commit() } + // Earlier specs in the same JVM can leave an appId behind. + beforeAny { + Logging.logLevel = LogLevel.NONE + clearPrefs() + } + + afterAny { + clearPrefs() + } + test("accessor instances after initWithContext without appID shows the failure reason") { // Given val context = getApplicationContext() diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt index c3ca1dffb..866a24656 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/OperationRepoTests.kt @@ -13,6 +13,7 @@ import com.onesignal.debug.internal.logging.Logging import com.onesignal.mocks.CoreInternalMocks import com.onesignal.mocks.MockHelper import com.onesignal.mocks.MockPreferencesService +import com.onesignal.session.internal.session.operations.SessionStateModelStore import com.onesignal.user.internal.jwt.JwtRequirement import com.onesignal.user.internal.jwt.JwtTokenStore import com.onesignal.user.internal.operations.CreateSubscriptionOperation @@ -94,6 +95,7 @@ private class Mocks { getNewRecordState(configModelStore), jwtTokenStore, identityVerificationService, + SessionStateModelStore(MockPreferencesService()), ), recordPrivateCalls = true, ) @@ -121,6 +123,7 @@ class OperationRepoTests : FunSpec({ getNewRecordState(mocks.configModelStore), JwtTokenStore(MockPreferencesService()), CoreInternalMocks.identityVerificationService(), + SessionStateModelStore(MockPreferencesService()), ), ) @@ -1068,6 +1071,8 @@ class OperationRepoTests : FunSpec({ // Then response shouldBe true + // waiters wake before the delay starts + coVerify(timeout = 1_000) { mocks.operationRepo.delayForPostCreate(any()) } coVerifyOrder { // ensure the order: IDs are translated, operation removed from the store, then delay for postCreateDelay operation.translateIds(idTranslation) @@ -1083,7 +1088,7 @@ class OperationRepoTests : FunSpec({ test("ensure each time enqueue is called it restarts the delay time") { // Given val mocks = Mocks() - mocks.configModelStore.model.opRepoExecutionInterval = 100 + mocks.configModelStore.model.opRepoExecutionInterval = 200 // When mocks.operationRepo.start() diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/SessionOperationQueueTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/SessionOperationQueueTests.kt new file mode 100644 index 000000000..9b662ad3a --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/operations/SessionOperationQueueTests.kt @@ -0,0 +1,438 @@ +package com.onesignal.core.internal.operations + +import com.onesignal.core.internal.operations.impl.OperationModelStore +import com.onesignal.core.internal.operations.impl.OperationRepo +import com.onesignal.core.internal.operations.impl.SessionQueuePolicy +import com.onesignal.debug.LogLevel +import com.onesignal.debug.internal.logging.Logging +import com.onesignal.mocks.CoreInternalMocks +import com.onesignal.mocks.MockHelper +import com.onesignal.mocks.MockPreferencesService +import com.onesignal.session.internal.session.operations.CreateSessionOperation +import com.onesignal.session.internal.session.operations.SessionStateModelStore +import com.onesignal.session.internal.session.operations.UpdateSessionOperation +import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor +import com.onesignal.user.internal.jwt.JwtRequirement +import com.onesignal.user.internal.jwt.JwtTokenStore +import com.onesignal.user.internal.operations.ExecutorMocks.Companion.getNewRecordState +import com.onesignal.user.internal.operations.SetTagOperation +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.collections.shouldContainExactly +import io.kotest.matchers.shouldBe +import io.mockk.coEvery +import io.mockk.every +import io.mockk.mockk +import io.mockk.spyk +import io.mockk.verify +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.delay +import kotlinx.coroutines.withTimeout + +private class SessionQueueMocks( + val prefs: MockPreferencesService = MockPreferencesService(), +) { + val configModelStore = + MockHelper.configModelStore { + it.isInitializedWithRemote = true + it.useIdentityVerification = JwtRequirement.NOT_REQUIRED + } + + val operationModelStore = OperationModelStore(prefs) + val sessionStateStore = SessionStateModelStore(prefs) + + val executed = mutableListOf() + var respond: (Operation) -> ExecutionResponse = { ExecutionResponse(ExecutionResult.SUCCESS) } + + val executor: IOperationExecutor = + mockk().also { + every { it.operations } returns listOf(SessionOperationExecutor.CREATE_SESSION, SessionOperationExecutor.UPDATE_SESSION) + coEvery { it.execute(any()) } answers { + val op = firstArg>().single() + executed.add(op) + respond(op) + } + } + + var identityVerificationService = CoreInternalMocks.identityVerificationService() + + val operationRepo: OperationRepo by lazy { + spyk( + OperationRepo( + listOf(executor), + operationModelStore, + configModelStore, + MockHelper.systemTime(), + getNewRecordState(configModelStore), + JwtTokenStore(MockPreferencesService()), + identityVerificationService, + sessionStateStore, + ), + recordPrivateCalls = true, + ) + } + + fun queued() = synchronized(operationRepo.queue) { operationRepo.queue.map { it.operation } } + + fun enqueueAll(vararg ops: Operation) { + ops.forEach { operationRepo.enqueue(it) } + verify(timeout = 1_000, exactly = ops.size) { + operationRepo["enqueueNew"](any(), any()) + } + } +} + +private fun create( + localSessionId: String = "local-s1", + onesignalId: String = "onesignalId", + externalId: String? = null, +) = CreateSessionOperation("appId", localSessionId, onesignalId, "subscriptionId", startTime = 1_000L, externalId) + +private fun update( + sessionId: String = "server-s1", + activeDuration: Long = 1_000L, + endTime: Long? = null, + onesignalId: String = "onesignalId", +) = UpdateSessionOperation("appId", sessionId, onesignalId, "subscriptionId", activeDuration, endTime) + +private fun item(op: Operation) = OperationRepo.OperationQueueItem(op, bucket = 0) + +private suspend fun awaitExecuted( + mocks: SessionQueueMocks, + count: Int, +) = withTimeout(5_000) { + while (mocks.executed.size < count) delay(10) +} + +class SessionOperationQueueTests : FunSpec({ + beforeAny { + Logging.logLevel = LogLevel.NONE + } + + test("duration-only updates collapse to the highest duration") { + val mocks = SessionQueueMocks() + + mocks.enqueueAll(update(activeDuration = 2_000L), update(activeDuration = 5_000L), update(activeDuration = 4_000L)) + + val queued = mocks.queued().single() as UpdateSessionOperation + queued.activeDuration shouldBe 5_000L + mocks.operationModelStore.list().map { it.id } shouldContainExactly listOf(queued.id) + } + + test("an end replaces an unsent update and keeps its duration") { + val mocks = SessionQueueMocks() + + mocks.enqueueAll(update(activeDuration = 5_000L), update(activeDuration = 3_000L, endTime = 9L)) + + val queued = mocks.queued().single() as UpdateSessionOperation + queued.endTime shouldBe 9L + queued.activeDuration shouldBe 5_000L + } + + test("an update after a queued end is dropped") { + val mocks = SessionQueueMocks() + + mocks.enqueueAll(update(endTime = 9L), update(activeDuration = 8_000L)) + + val queued = mocks.queued().single() as UpdateSessionOperation + queued.endTime shouldBe 9L + } + + test("updates for different sessions are not coalesced") { + val mocks = SessionQueueMocks() + + mocks.enqueueAll(update(sessionId = "s1"), update(sessionId = "s2")) + + mocks.queued().size shouldBe 2 + } + + test("cap drops the oldest session with all its operations") { + val mocks = SessionQueueMocks() + val max = SessionQueuePolicy.MAX_QUEUED_SESSION_OPERATIONS + val others = (1 until max).map { create(localSessionId = "local-other-$it") } + + mocks.enqueueAll(create(), update(sessionId = "local-s1"), *others.toTypedArray()) + + mocks.queued() shouldContainExactly others + mocks.operationModelStore.list().size shouldBe others.size + } + + test("non-session operations do not count toward the cap") { + val policy = SessionQueuePolicy(maxQueued = 2) + val queue = mutableListOf(item(create(localSessionId = "local-a")), item(create(localSessionId = "local-b"))) + + val result = policy.onEnqueue(queue, item(SetTagOperation("appId", "onesignalId", null, "key", "value"))) + + result.accept shouldBe true + result.removed shouldBe emptyList() + queue.size shouldBe 2 + } + + test("a coalesced update keeps the backoff of the update it replaces") { + val policy = SessionQueuePolicy() + val backedOff = item(update()).apply { + retries = 3 + retryAt = 60_000L + } + val incoming = item(update(activeDuration = 2_000L)) + + policy.onEnqueue(mutableListOf(backedOff), incoming) + + incoming.retries shouldBe 3 + incoming.retryAt shouldBe 60_000L + } + + test("an update after an end that is executing is dropped") { + val policy = SessionQueuePolicy() + val queue = mutableListOf() + policy.onEnqueue(queue, item(update(endTime = 9L))).accept shouldBe true + + // The end leaves the queue while it executes. + policy.onEnqueue(queue, item(update(activeDuration = 8_000L))).accept shouldBe false + } + + test("an end recorded under the local ID still drops updates after the create succeeds") { + val policy = SessionQueuePolicy() + val queue = mutableListOf() + policy.onEnqueue(queue, item(create())) + policy.onEnqueue(queue, item(update(sessionId = "local-s1", endTime = 9L))).accept shouldBe true + + policy.recordTranslations(mapOf("local-s1" to "server-s1")) + + policy.onEnqueue(mutableListOf(), item(update(sessionId = "server-s1"))).accept shouldBe false + } + + test("an update for a session whose create was dropped is rejected") { + val policy = SessionQueuePolicy() + + policy.onDropped(mutableListOf(), create()) + + policy.onEnqueue(mutableListOf(), item(update(sessionId = "local-s1"))).accept shouldBe false + } + + test("an update with a local session ID and no create is rejected") { + SessionQueuePolicy().onEnqueue(mutableListOf(), item(update(sessionId = "local-s1"))).accept shouldBe false + } + + test("an update carries the IDs its create was sent with") { + val policy = SessionQueuePolicy() + val queue = mutableListOf() + val create = create(onesignalId = "local-user") + policy.onEnqueue(queue, item(create)) + val queued = update(sessionId = "local-s1", onesignalId = "user-b") + + policy.onEnqueue(queue, item(queued)) + + queued.onesignalId shouldBe "local-user" + } + + test("an update after its create succeeded carries the create's IDs") { + val policy = SessionQueuePolicy() + val create = create(onesignalId = "user-a") + policy.onEnqueue(mutableListOf(), item(create)) + policy.recordTranslations(mapOf("local-s1" to "server-s1")) + val later = update(sessionId = "server-s1", onesignalId = "user-b") + + policy.onEnqueue(mutableListOf(), item(later)) + + later.onesignalId shouldBe "user-a" + } + + test("an update after a restart carries the IDs of a create from before it") { + val prefs = MockPreferencesService() + SessionQueuePolicy(sessions = SessionStateModelStore(prefs)).onEnqueue(mutableListOf(), item(create(onesignalId = "user-a"))) + val later = update(sessionId = "local-s1", onesignalId = "user-b") + + val accepted = SessionQueuePolicy(sessions = SessionStateModelStore(prefs)).onEnqueue(mutableListOf(), item(later)).accept + + accepted shouldBe true + later.onesignalId shouldBe "user-a" + } + + test("an update after an end that already left the queue is rejected after a restart") { + val prefs = MockPreferencesService() + SessionQueuePolicy(sessions = SessionStateModelStore(prefs)).onEnqueue(mutableListOf(), item(update(endTime = 2_000L))) + + SessionQueuePolicy(sessions = SessionStateModelStore(prefs)).onEnqueue(mutableListOf(), item(update())).accept shouldBe false + } + + test("an update after a restart is sent with the backend session ID of a create from before it") { + val prefs = MockPreferencesService() + SessionQueuePolicy(sessions = SessionStateModelStore(prefs)).apply { + onEnqueue(mutableListOf(), item(create())) + recordTranslations(mapOf("local-s1" to "server-s1")) + } + val later = update(sessionId = "local-s1") + + SessionQueuePolicy(sessions = SessionStateModelStore(prefs)).onEnqueue(mutableListOf(), item(later)) + + later.sessionId shouldBe "server-s1" + } + + test("the session state cap never drops a session with operations still queued") { + val policy = SessionQueuePolicy(maxQueued = 2) + val queue = mutableListOf() + val queued = item(create(localSessionId = "local-a")) + policy.onEnqueue(queue, queued) + queue.add(queued) + + policy.onEnqueue(queue, item(create(localSessionId = "local-b"))) + policy.onEnqueue(queue, item(create(localSessionId = "local-c"))) + + policy.onEnqueue(queue, item(update(sessionId = "local-a"))).accept shouldBe true + policy.onEnqueue(queue, item(update(sessionId = "local-b"))).accept shouldBe false + } + + test("session operations are not stored before privacy consent") { + val mocks = SessionQueueMocks() + mocks.configModelStore.model.consentRequired = true + mocks.configModelStore.model.consentGiven = false + + mocks.enqueueAll(create(), update()) + + mocks.queued() shouldBe emptyList() + mocks.operationModelStore.list() shouldBe emptyList() + } + + test("an update for a session whose create was rejected is rejected") { + var awaitingConsent = true + val policy = SessionQueuePolicy(awaitingConsent = { awaitingConsent }) + policy.onEnqueue(mutableListOf(), item(create())).accept shouldBe false + + awaitingConsent = false + + policy.onEnqueue(mutableListOf(), item(update(sessionId = "local-s1"))).accept shouldBe false + } + + test("a session create does not stall other operations for the post-create delay") { + val mocks = SessionQueueMocks() + mocks.configModelStore.model.opRepoPostCreateDelay = 60_000L + mocks.respond = { op -> + if (op is CreateSessionOperation) { + ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(op.localSessionId to "server-s1")) + } else { + ExecutionResponse(ExecutionResult.SUCCESS) + } + } + + mocks.enqueueAll(create(), update(sessionId = "server-other")) + mocks.operationRepo.start() + awaitExecuted(mocks, 2) + + (mocks.executed[1] as UpdateSessionOperation).sessionId shouldBe "server-other" + } + + test("an update waits for its create and is sent with the backend session ID") { + val mocks = SessionQueueMocks() + mocks.respond = { op -> + if (op is CreateSessionOperation) { + ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(op.localSessionId to "server-s1")) + } else { + ExecutionResponse(ExecutionResult.SUCCESS) + } + } + + mocks.enqueueAll(create(), update(sessionId = "local-s1")) + mocks.operationRepo.start() + awaitExecuted(mocks, 2) + + mocks.executed[0].shouldBeCreate() + (mocks.executed[1] as UpdateSessionOperation).sessionId shouldBe "server-s1" + } + + test("an update built from the local ID after its create succeeded is still sent") { + val mocks = SessionQueueMocks() + val created = CompletableDeferred() + mocks.respond = { op -> + if (op is CreateSessionOperation) { + created.complete(Unit) + ExecutionResponse(ExecutionResult.SUCCESS, idTranslations = mapOf(op.localSessionId to "server-s1")) + } else { + ExecutionResponse(ExecutionResult.SUCCESS) + } + } + mocks.operationRepo.start() + mocks.operationRepo.enqueue(create()) + created.await() + awaitExecuted(mocks, 1) + + mocks.operationRepo.enqueue(update(sessionId = "local-s1")) + awaitExecuted(mocks, 2) + + (mocks.executed[1] as UpdateSessionOperation).sessionId shouldBe "server-s1" + } + + test("a dropped create drops its queued updates") { + val mocks = SessionQueueMocks() + mocks.respond = { ExecutionResponse(ExecutionResult.FAIL_NORETRY) } + + mocks.enqueueAll(create(), update(sessionId = "local-s1"), update(sessionId = "server-other")) + mocks.operationRepo.start() + awaitExecuted(mocks, 2) + + mocks.executed.map { it.applyToRecordId } shouldContainExactly listOf("local-s1", "server-other") + mocks.queued() shouldBe emptyList() + mocks.operationModelStore.list() shouldBe emptyList() + } + + test("a create waits for a backend onesignalId") { + val op = create(onesignalId = "local-user") + op.canStartExecute shouldBe false + + op.translateIds(mapOf("local-user" to "user")) + + op.canStartExecute shouldBe true + op.onesignalId shouldBe "user" + } + + test("a persisted create replays on the next launch with the same idempotency key") { + // Given + val prefs = MockPreferencesService() + val op = create().apply { id = "op-id" } + OperationModelStore(prefs).apply { loadOperations() }.add(op) + + // When + val mocks = SessionQueueMocks(prefs) + mocks.operationRepo.start() + awaitExecuted(mocks, 1) + + // Then + (mocks.executed.single() as CreateSessionOperation).idempotencyKey shouldBe op.idempotencyKey + } + + test("session operations are kept for anonymous users when identity verification is required") { + val mocks = SessionQueueMocks() + mocks.configModelStore.model.useIdentityVerification = JwtRequirement.REQUIRED + mocks.identityVerificationService = CoreInternalMocks.identityVerificationService(newCodePathsRun = true, ivBehaviorActive = true) + + mocks.enqueueAll(create()) + mocks.operationRepo.removeOperationsWithoutExternalId() + + mocks.queued().size shouldBe 1 + } + + test("a create for an anonymous user that was never created is purged with its updates under identity verification") { + val mocks = SessionQueueMocks() + mocks.identityVerificationService = CoreInternalMocks.identityVerificationService(newCodePathsRun = true, ivBehaviorActive = true) + val kept = create(localSessionId = "local-s2", onesignalId = "local-user", externalId = "alice") + + mocks.enqueueAll(create(onesignalId = "local-anon"), update(sessionId = "local-s1"), kept) + mocks.configModelStore.model.useIdentityVerification = JwtRequirement.REQUIRED + mocks.operationRepo.removeOperationsWithoutExternalId() + + mocks.queued() shouldContainExactly listOf(kept) + mocks.operationModelStore.list() shouldContainExactly listOf(kept) + } + + test("a create for an anonymous user that was never created is not enqueued under identity verification") { + val mocks = SessionQueueMocks() + mocks.configModelStore.model.useIdentityVerification = JwtRequirement.REQUIRED + mocks.identityVerificationService = CoreInternalMocks.identityVerificationService(newCodePathsRun = true, ivBehaviorActive = true) + + mocks.operationRepo.enqueueAndAwaitResult(create(onesignalId = "local-anon")).success shouldBe false + + mocks.queued() shouldBe emptyList() + mocks.operationModelStore.list() shouldBe emptyList() + } +}) + +private fun Operation.shouldBeCreate() = (this is CreateSessionOperation) shouldBe true diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/SessionServiceTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/SessionServiceTests.kt index 746653d72..49e564867 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/SessionServiceTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/SessionServiceTests.kt @@ -1,5 +1,6 @@ package com.onesignal.session.internal.session +import com.onesignal.common.IDManager import com.onesignal.common.modeling.ISingletonModelStoreChangeHandler import com.onesignal.common.modeling.ModelChangeTags import com.onesignal.common.modeling.ModelChangedArgs @@ -17,6 +18,7 @@ import com.onesignal.user.internal.backend.IdentityConstants import com.onesignal.user.internal.identity.IdentityModel import io.kotest.core.spec.style.FunSpec import io.kotest.matchers.shouldBe +import io.kotest.matchers.shouldNotBe import io.mockk.every import io.mockk.just import io.mockk.mockk @@ -300,7 +302,7 @@ class SessionServiceTests : FunSpec({ sessionModelStore.model.usesSessionsApi shouldBe true } - test("new session pins onesignal and subscription IDs and clears the server session ID") { + test("new session pins onesignal and subscription IDs and a new local session ID") { // Given val mocks = Mocks() val sessionService = mocks.sessionService @@ -309,7 +311,7 @@ class SessionServiceTests : FunSpec({ val sessionModelStore = mocks.sessionModelStore { it.isValid = false - it.serverSessionId = "previous-server-session" + it.localSessionId = "local-previous" } // When @@ -318,7 +320,8 @@ class SessionServiceTests : FunSpec({ // Then sessionModelStore.model.onesignalId shouldBe Mocks.ONESIGNAL_ID sessionModelStore.model.subscriptionId shouldBe Mocks.SUBSCRIPTION_ID - sessionModelStore.model.serverSessionId shouldBe null + sessionModelStore.model.localSessionId shouldNotBe "local-previous" + IDManager.isLocalId(sessionModelStore.model.localSessionId) shouldBe true } test("pinned IDs stay the same after login within a session") { @@ -508,7 +511,6 @@ class SessionServiceTests : FunSpec({ store.model.usesSessionsApi = true store.model.onesignalId = Mocks.ONESIGNAL_ID store.model.subscriptionId = Mocks.SUBSCRIPTION_ID - store.model.serverSessionId = "server-session-id" // When val reloaded = SessionModelStore(prefs).model @@ -521,7 +523,6 @@ class SessionServiceTests : FunSpec({ reloaded.usesSessionsApi shouldBe true reloaded.onesignalId shouldBe Mocks.ONESIGNAL_ID reloaded.subscriptionId shouldBe Mocks.SUBSCRIPTION_ID - reloaded.serverSessionId shouldBe "server-session-id" } test("onFocus dispatches the session-mutation body through runOnSerialIO (SDK-4508)") { diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt new file mode 100644 index 000000000..811977702 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/operations/SessionOperationExecutorTests.kt @@ -0,0 +1,155 @@ +package com.onesignal.session.internal.session.operations + +import com.onesignal.common.exceptions.BackendException +import com.onesignal.core.internal.operations.ExecutionResult +import com.onesignal.core.internal.operations.impl.OperationModelStore +import com.onesignal.debug.LogLevel +import com.onesignal.debug.internal.logging.Logging +import com.onesignal.mocks.MockHelper +import com.onesignal.mocks.MockPreferencesService +import com.onesignal.session.internal.session.backend.CreateSessionRequest +import com.onesignal.session.internal.session.backend.ISessionsBackendService +import com.onesignal.session.internal.session.backend.UpdateSessionRequest +import com.onesignal.session.internal.session.operations.impl.SessionOperationExecutor +import com.onesignal.user.internal.operations.impl.states.NewRecordsState +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.shouldBe +import io.kotest.matchers.types.shouldBeInstanceOf +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.every +import io.mockk.mockk + +private class Mocks { + val backend = mockk() + val newRecordState = mockk().also { every { it.isInMissingRetryWindow(any()) } returns false } + val executor = + SessionOperationExecutor( + backend, + MockHelper.deviceService(), + newRecordState, + ) +} + +private fun create() = CreateSessionOperation("appId", "local-session", "onesignalId", "subscriptionId", startTime = 1_700_000_000_500L, externalId = null) + +private fun update(endTime: Long? = null) = + UpdateSessionOperation("appId", "server-session", "onesignalId", "subscriptionId", activeDuration = 61_900L, endTime = endTime) + +class SessionOperationExecutorTests : FunSpec({ + beforeAny { + Logging.logLevel = LogLevel.NONE + } + + test("create sends the request and translates the local session ID") { + // Given + val mocks = Mocks() + val op = create() + coEvery { mocks.backend.createSession(any(), any()) } returns "server-session" + + // When + val response = mocks.executor.execute(listOf(op)) + + // Then + response.result shouldBe ExecutionResult.SUCCESS + response.idTranslations shouldBe mapOf("local-session" to "server-session") + coVerify { + mocks.backend.createSession( + "appId", + CreateSessionRequest("onesignalId", "subscriptionId", 1, 1_700_000_000L, op.idempotencyKey), + ) + } + } + + test("create without a session_id in the response is retried, then dropped") { + val mocks = Mocks() + val op = create() + coEvery { mocks.backend.createSession(any(), any()) } returns null + + val results = (1..SessionOperationExecutor.MAX_MISSING_SESSION_ID_ATTEMPTS).map { mocks.executor.execute(listOf(op)).result } + + results shouldBe listOf(ExecutionResult.FAIL_RETRY, ExecutionResult.FAIL_RETRY, ExecutionResult.FAIL_NORETRY) + coVerify(exactly = 3) { mocks.backend.createSession(any(), match { it.idempotencyKey == op.idempotencyKey }) } + } + + test("update sends the duration and end time in seconds") { + val mocks = Mocks() + val op = update(endTime = 1_700_000_100_999L) + coEvery { mocks.backend.updateSession(any(), any(), any()) } returns Unit + + mocks.executor.execute(listOf(op)).result shouldBe ExecutionResult.SUCCESS + + coVerify { + mocks.backend.updateSession( + "appId", + "server-session", + UpdateSessionRequest("onesignalId", "subscriptionId", 61L, op.idempotencyKey, 1_700_000_100L), + ) + } + } + + for (status in listOf(-1, 0, 408, 429, 500, 503)) { + test("status $status is retried with its Retry-After") { + val mocks = Mocks() + coEvery { mocks.backend.createSession(any(), any()) } throws BackendException(status, retryAfterSeconds = 7) + + val response = mocks.executor.execute(listOf(create())) + + response.result shouldBe ExecutionResult.FAIL_RETRY + response.retryAfterSeconds shouldBe 7 + } + } + + for (status in listOf(400, 401, 403, 404, 409, 422)) { + test("status $status is dropped") { + val mocks = Mocks() + coEvery { mocks.backend.updateSession(any(), any(), any()) } throws BackendException(status) + + mocks.executor.execute(listOf(update())).result shouldBe ExecutionResult.FAIL_NORETRY + } + } + + test("update 404 right after the create is retried") { + val mocks = Mocks() + every { mocks.newRecordState.isInMissingRetryWindow("server-session") } returns true + coEvery { mocks.backend.updateSession(any(), any(), any()) } throws BackendException(404) + + mocks.executor.execute(listOf(update())).result shouldBe ExecutionResult.FAIL_RETRY + } + + test("create 404 while the user is new is retried") { + val mocks = Mocks() + every { mocks.newRecordState.isInMissingRetryWindow("onesignalId") } returns true + coEvery { mocks.backend.createSession(any(), any()) } throws BackendException(404) + + mocks.executor.execute(listOf(create())).result shouldBe ExecutionResult.FAIL_RETRY + } + + test("idempotency keys survive persistence") { + // Given + val store = OperationModelStore(MockPreferencesService()) + val create = create() + val update = update(endTime = 1L) + + // When + val restoredCreate = store.create(create.toJSON()) + val restoredUpdate = store.create(update.toJSON()) + + // Then + restoredCreate.shouldBeInstanceOf() + restoredCreate.idempotencyKey shouldBe create.idempotencyKey + restoredCreate.startTime shouldBe create.startTime + restoredCreate.missingSessionIdAttempts shouldBe 0 + restoredUpdate.shouldBeInstanceOf() + restoredUpdate.idempotencyKey shouldBe update.idempotencyKey + restoredUpdate.endTime shouldBe 1L + } + + test("a create is keyed by its local session ID") { + create().idempotencyKey shouldBe "local-session" + } + + test("each update gets its own idempotency key") { + (update().idempotencyKey == update().idempotencyKey) shouldBe false + } +}) diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt index 52ed67075..5f32d7911 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/user/internal/migrations/RecoverFromDroppedLoginBugTests.kt @@ -7,6 +7,7 @@ import com.onesignal.debug.internal.logging.Logging import com.onesignal.mocks.CoreInternalMocks import com.onesignal.mocks.MockHelper import com.onesignal.mocks.MockPreferencesService +import com.onesignal.session.internal.session.operations.SessionStateModelStore import com.onesignal.user.internal.jwt.JwtRequirement import com.onesignal.user.internal.jwt.JwtTokenStore import com.onesignal.user.internal.operations.ExecutorMocks @@ -47,6 +48,7 @@ private class Mocks { ExecutorMocks.getNewRecordState(configModelStore), JwtTokenStore(MockPreferencesService()), CoreInternalMocks.identityVerificationService(), + SessionStateModelStore(MockPreferencesService()), ), )