diff --git a/CHANGELOG.md b/CHANGELOG.md index a74749a8..d92b494e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,25 @@ # Change Log +## 27.1.0-rc.7 + +* Fixed: `host()` reports the live connection state instead of always resolving open +* Fixed: a disconnect event now fires when a retired or dropped background connection closes +* Fixed: notification taps are recorded after launching the app, so tap-opened screens land on top +* Fixed: subscribing on a `Push` closed mid-connect now throws instead of silently proceeding + +## 27.1.0-rc.6 + +* Added: `getInitialNotification()`, `onNotificationOpened()`, and the `PushNotificationOpened` type (topic + data) +* Fixed: Android `onOpen`/`onClose` now fire for `background` subscriptions +* Fixed: `subscribe` falls back to the session the client signed in with +* Fixed: Android notifications no longer suppressed when no live callback received the message +* Fixed: `close()` during subscribe/resume no longer re-subscribes; `resume` leaves a live `Push` host untouched +* Updated: notification taps routed through a translucent `PushOpenActivity` + +## 27.1.0-rc.5 + +* Updated: regenerated with sdk-generator 5.5.0; no SDK code changes + ## 27.1.0-rc.3 * Added: background push notifications render the server `notification` title, body, and image diff --git a/README.md b/README.md index f8d33ae5..a3c835ba 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,7 @@ Add this to your package's `pubspec.yaml` file: ```yml dependencies: - appwrite: ^27.1.0-rc.3 + appwrite: ^27.1.0-rc.7 ``` You can install packages from the command line: @@ -53,8 +53,8 @@ dependencies { A subscription with `background: true` keeps delivering after the app is backgrounded, killed or the device restarts, until it is unsubscribed or `push.close()` is called (do this on sign-out). The plugin saves the subscription, and a scheduled job and alarm wake the app every 15 to 60 -seconds to reconnect; the broker replays what was sent in between (`retry: true`). Messages no -in-app callback receives are posted as notifications that open the app. It reconnects with the +seconds to reconnect; the broker replays what was sent in between (`retry: true`). While the app +is not on screen, each message is posted as a notification that opens the app. It reconnects with the credential saved at subscribe time, so use a session rather than a short-lived JWT. On Android 13 and later, the first background subscription asks the user for the @@ -62,6 +62,10 @@ On Android 13 and later, the first background subscription asks the user for the callback but posts no notification. Notifications show the title, body and image sent with `createPush`, and fall back to the subscription's `title` and the raw payload for other messages. +Notifications are posted while the app is backgrounded or closed. While it is on screen your +callback shows the message, so none is posted unless the subscription passes +`notifyInForeground: true`. + ```dart final sub = await push.subscribe('news', (message) => print(message.data), background: true, title: 'News'); @@ -76,6 +80,59 @@ declare in the Play Console. Apps that never enable it can remove the service fr manifest with `tools:node="remove"` on `io.appwrite.services.PushService` and `android.permission.FOREGROUND_SERVICE_REMOTE_MESSAGING`. +#### Delivery while the app is closed + +Messages sent while the app is closed arrive at the next scheduled wake-up. While the device is +awake that is about every 15 seconds, or about every 60 seconds once exact alarms are allowed; +without exact alarms the wake-ups are inexact, so battery saver can defer them further. In Doze +(screen off and idle for a while) Android limits background alarms, exact ones included, to about +one every nine minutes, so a closed app can take several minutes to receive a message: allowing +exact alarms makes wake-ups punctual, it does not lift Doze. For immediate delivery, also in +Doze, use `push.setForeground(true)` (see above). + +The SDK uses exact alarms on its own whenever the app may schedule them. To allow it: + +1. Declare the permissions in your app's `AndroidManifest.xml`. Both are optional and subject to + Google Play policy: `SCHEDULE_EXACT_ALARM` needs a declaration in the Play Console, and + `USE_EXACT_ALARM` is reserved for alarm, clock and calendar apps (the SDK does not use it). + + ```xml + + + ``` + +2. On Android 13 and later the user has to allow exact alarms, under Settings > Apps > Special app + access > Alarms & reminders. Android 12 grants a declared `SCHEDULE_EXACT_ALARM` + automatically, and older versions need nothing. Check with `await push.backgroundStatus()`: when `bestEffort` is + true, explain why to the user, then from a user action open that screen with `push.requestExactAlarms()`, or + ask for the battery-optimisation exemption with `push.requestIgnoreBatteryOptimizations()`. Both return false when there is + nothing to ask, including when the permission is not declared. The SDK never opens these + screens on its own. + +If the user force-stops the app (Settings > Force stop, and on some devices swiping it away from +recents), Android cancels its alarms and jobs: nothing is delivered until the app is opened +again, and the broker then replays what was sent meanwhile. + +Set the notification icon with +`` in your +``; without it a generic icon is used. + +### Opening a tapped notification + +Read the `data` sent with `createPush` when the user taps a background notification: + +```dart +// The tap that launched the app (reported once, so call it at startup). +final opened = await push.getInitialNotification(); +if (opened != null) openSale(opened.data['saleId']); + +// Taps while the app is running, including in the background. +final stop = push.onNotificationOpened((opened) => openSale(opened.data['saleId'])); +``` + +On Android the SDK's native plugin reports the taps, on iOS `flutter_local_notifications`, and +on the web the notification's click while the page is open. + ## Getting Started diff --git a/android/build.gradle b/android/build.gradle index 6770cc69..8e542eb9 100644 --- a/android/build.gradle +++ b/android/build.gradle @@ -1,7 +1,7 @@ // The native half of Push on Android: background delivery that survives the process being // killed, a reboot and an app update, shared with the Appwrite Android SDK. group = "io.appwrite.flutter" -version = "27.1.0-rc.3" +version = "27.1.0-rc.7" buildscript { ext.kotlin_version = "2.1.0" @@ -28,7 +28,7 @@ apply plugin: "kotlin-android" android { namespace = "io.appwrite.flutter" - compileSdk = 35 + compileSdk = 36 compileOptions { sourceCompatibility = JavaVersion.VERSION_17 diff --git a/android/src/main/AndroidManifest.xml b/android/src/main/AndroidManifest.xml index 44f894ed..d2b7d500 100644 --- a/android/src/main/AndroidManifest.xml +++ b/android/src/main/AndroidManifest.xml @@ -16,6 +16,13 @@ + Unit)? = null private var bridge: PushBridge? = null private var activity: Activity? = null @@ -49,6 +52,8 @@ class AppwritePushPlugin : ) override fun onError(message: String) = send(mapOf("type" to "error", "message" to message)) + + override fun onConnection(connected: Boolean) = send(mapOf("type" to "connection", "connected" to connected)) }, ) methods = MethodChannel(binding.binaryMessenger, METHOD_CHANNEL).also { it.setMethodCallHandler(this) } @@ -61,6 +66,9 @@ class AppwritePushPlugin : methods = null events = null sink = null + stopOpened?.invoke() + stopOpened = null + openedListeners = 0 } override fun onAttachedToActivity(binding: ActivityPluginBinding) { @@ -87,7 +95,7 @@ class AppwritePushPlugin : bridge.host(call.argument("config")!!, call.argument("subscriptions")!!) { error -> main.post { if (error == null) { - result.success(null) + result.success(bridge.isConnected()) } else { result.error(ERROR_CODE, error, null) } @@ -103,9 +111,14 @@ class AppwritePushPlugin : "stop" -> bridge.stop().let { null } "setForeground" -> bridge.setForeground(call.argument("enabled") == true).let { null } "hasSaved" -> bridge.hasSaved() - "resume" -> bridge.resume().let { null } + "resume" -> bridge.resume(call.argument("authMethod"), call.argument("credential"), false).let { null } + "backgroundStatus" -> bridge.backgroundStatus() + "requestExactAlarms" -> bridge.requestExactAlarms() + "requestIgnoreBatteryOptimizations" -> bridge.requestIgnoreBatteryOptimizations() "setErrorCallback" -> bridge.setErrorCallback(call.argument("registered") == true) "defaultClientId" -> bridge.defaultClientId(call.argument("authMethod")!!, call.argument("credential")!!) + "getInitialNotification" -> PushTaps.take()?.let { mapOf("topic" to it.topic, "payload" to it.payload) } + "listenOpened" -> listenOpened(call.argument("listening") == true).let { null } else -> return result.notImplemented() }, ) @@ -140,6 +153,17 @@ class AppwritePushPlugin : main.post { sink?.success(event) } } + // Sends each tap while Dart listens for them; until then a tap waits for getInitialNotification. + private fun listenOpened(listening: Boolean) { + openedListeners = (openedListeners + if (listening) 1 else -1).coerceAtLeast(0) + if (openedListeners > 0 && stopOpened == null) { + stopOpened = PushTaps.listen { tap -> send(mapOf("type" to "opened", "topic" to tap.topic, "payload" to tap.payload)) } + } else if (openedListeners == 0) { + stopOpened?.invoke() + stopOpened = null + } + } + private companion object { const val METHOD_CHANNEL = "appwrite.push" const val EVENT_CHANNEL = "appwrite.push/events" diff --git a/android/src/main/kotlin/io/appwrite/services/PushBackground.kt b/android/src/main/kotlin/io/appwrite/services/PushBackground.kt index a550a555..e06dfabe 100644 --- a/android/src/main/kotlin/io/appwrite/services/PushBackground.kt +++ b/android/src/main/kotlin/io/appwrite/services/PushBackground.kt @@ -1,5 +1,7 @@ package io.appwrite.services +import android.Manifest +import android.app.ActivityManager import android.app.AlarmManager import android.app.NotificationChannel import android.app.NotificationManager @@ -14,11 +16,14 @@ import android.graphics.Bitmap import android.graphics.BitmapFactory import android.net.ConnectivityManager import android.net.Network +import android.net.Uri import android.os.Build import android.os.PowerManager import android.os.SystemClock +import android.provider.Settings import android.util.AtomicFile import android.util.Log +import android.webkit.CookieManager import androidx.core.app.NotificationCompat import androidx.core.app.NotificationManagerCompat import androidx.core.content.ContextCompat @@ -30,8 +35,10 @@ import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish import io.appwrite.exceptions.AppwriteException import org.json.JSONArray import org.json.JSONObject +import java.io.ByteArrayOutputStream import java.io.File import java.io.IOException +import java.io.InputStream import java.net.HttpURLConnection import java.net.URL import java.util.UUID @@ -53,6 +60,7 @@ internal data class PushEntry( val filter: String, val retry: Boolean, val title: String?, + val notifyInForeground: Boolean = false, ) /** @@ -96,7 +104,11 @@ internal object PushBackground { const val EXTRA_PAYLOAD = "io.appwrite.push.PAYLOAD" private const val SERVICE_CHANNEL_ID = "appwrite-push-service" - private const val JOB_ID = 0x50555348 + const val JOB_ID = 0x50555348 + const val NEXT_JOB_ID = 0x5055534A + const val EXPEDITED_JOB_ID = 0x5055534B + const val WATCHDOG_JOB_ID = 0x50555349 + private const val WATCHDOG_INTERVAL_MS = 15 * 60 * 1_000L private const val INTERVAL_MS = 15_000L private const val PRIVILEGED_INTERVAL_MS = 60_000L private const val RESTART_DELAY_MS = 3_000L @@ -116,6 +128,15 @@ internal object PushBackground { private const val HEARTBEAT_RESET_MS = 3 * 24 * 60 * 60 * 1_000L private const val REQUEST_TIMEOUT_SECONDS = 10L private const val IMAGE_TIMEOUT_MS = 5_000 + + // How long a scheduled run stays up after connecting so the broker can replay what was missed: + // until nothing arrived for DRAIN_QUIET_MS and nothing is being delivered, at most the run's cap. + const val JOB_DRAIN_MS = 15_000L + const val RECEIVER_DRAIN_MS = 5_000L + private const val DRAIN_QUIET_MS = 2_000L + private const val DRAIN_POLL_MS = 200L + private const val IMAGE_MAX_BYTES = 5 * 1024 * 1024 + private const val IMAGE_MAX_PX = 1_024 private const val CONNECT_TIMEOUT_SECONDS = 20L // How long a message waits for a listener that acknowledges it itself (the React Native and @@ -139,6 +160,16 @@ internal object PushBackground { @Volatile var errorHandler: ((Throwable) -> Boolean)? = null + // Tells the Push instance that hosts its subscriptions here when the connection opens (true) + // or closes (false). + @Volatile + var connectionHandler: ((Boolean) -> Unit)? = null + + // Whether the live connection (not a retired one) is open. + @Volatile + var isConnected = false + private set + @Volatile var serviceRunning = false @@ -164,6 +195,15 @@ internal object PushBackground { Thread(runnable, "AppwritePushBackground").apply { isDaemon = true } } private var mqtt: Mqtt5AsyncClient? = null + + // Set when [mqtt] is replaced or closed, so that client stops reconnecting and delivering. + private var mqttRetired: AtomicBoolean? = null + + // Messages being delivered, and when one last arrived or finished, for a run's drain window. + private val delivering = AtomicInteger(0) + + @Volatile + private var lastDelivery = 0L private var connectedConfig: PushConfig? = null private var connectedNetwork: Network? = null private var lastHeartbeat = 0L @@ -264,23 +304,120 @@ internal object PushBackground { } } - /** Resume saved background delivery when the app starts, without waiting for the next run. */ - fun resume(context: Context) { + /** + * Resume saved background delivery when the app starts, without waiting for the next run, with + * the app's current credential. A rotated session of the same user replaces the saved one; a + * different user drops the saved subscriptions, as does no credential when + * [signedOutWhenMissing] (the caller read the session store, so none means signed out). + * While a live Push hosts subscriptions here, it does nothing: that Push owns the connection. + */ + fun resume(context: Context, authMethod: String? = null, credential: String? = null, signedOutWhenMissing: Boolean = false) { val app = context.applicationContext - val active = synchronized(lock) { + val resume = synchronized(lock) { load(app) - isActive() + val current = config + when { + !isActive() || current == null -> null + // A live Push hosts here and owns the connection and its credential; resuming only + // reconciles saved delivery, so it leaves a live host alone. + listeners.isNotEmpty() -> null + // No credential: signed out when the caller can tell, else possibly not set yet. + authMethod == null || credential == null -> if (signedOutWhenMissing) false else null + current.authMethod == authMethod && current.credential == credential -> true + // A new credential for the same user (a rotated session) replaces the saved one, + // so the stale one is never sent and refused. + sameUser(current.authMethod, current.credential, authMethod, credential) -> { + config = current.copy(authMethod = authMethod, credential = credential) + persist(app) + true + } + // Another user signed in: the saved subscriptions were not theirs. + else -> false + } } - if (active) { - tick(app) {} + when (resume) { + true -> tick(app) {} + false -> stop(app) + null -> Unit } } + private enum class CredentialRefresh { UNCHANGED, CHANGED, GONE } + + // Bring the saved credential up to date with the session cookie the app + // signs in with, when it comes from one (see [PushConfig.sessionCookieUrl]): a rotated session of + // the same user replaces it, and no session or another user's is GONE. UNCHANGED when there is + // no cookie source or it cannot be read. + private fun refreshCredential(context: Context): CredentialRefresh { + val current = synchronized(lock) { config } ?: return CredentialRefresh.UNCHANGED + val url = current.sessionCookieUrl ?: return CredentialRefresh.UNCHANGED + val session = readSessionCookie(url, current.project) ?: return CredentialRefresh.UNCHANGED + return when { + session.isEmpty() -> CredentialRefresh.GONE + current.authMethod == "appwrite-session" && session == current.credential -> CredentialRefresh.UNCHANGED + sameUser(current.authMethod, current.credential, "appwrite-session", session) -> { + synchronized(lock) { + config = current.copy(authMethod = "appwrite-session", credential = session) + persist(context.applicationContext) + } + CredentialRefresh.CHANGED + } + else -> CredentialRefresh.GONE + } + } + + // The credential to send for a connection made with [connected]: the session re-read from the + // cookie store when one backs it (so an automatic reconnect uses a rotated session), else the + // one it connected with. Only a newer credential of the same user, for the same connection + // (broker, client id, project), is borrowed: a connection being replaced by [host] keeps its + // own. Null when the app signed out or another user signed in: authentication is aborted and + // delivery stops. + private fun currentCredential(context: Context, connected: PushConfig): String? { + if (refreshCredential(context) == CredentialRefresh.GONE) { + worker.execute { stop(context) } + return null + } + val current = synchronized(lock) { config } ?: return connected.credential + val sameConnection = current.host == connected.host && current.port == connected.port && + current.clientId == connected.clientId && current.project == connected.project && + current.authMethod == connected.authMethod + return if (sameConnection && sameUser(connected.authMethod, connected.credential, current.authMethod, current.credential)) { + current.credential + } else { + connected.credential + } + } + + // The `a_session_` cookie for [url] in the WebView cookie store, decoded; "" when there + // is none, and null when the store cannot be read. + private fun readSessionCookie(url: String, project: String): String? = runCatching { + val name = "a_session_$project" + CookieManager.getInstance().getCookie(url) + ?.split(";") + ?.map { it.trim() } + ?.firstOrNull { it.substringBefore("=") == name } + ?.substringAfter("=") + ?.let { Uri.decode(it) } + ?: "" + }.getOrNull() + + private fun sameUser(savedMethod: String, saved: String, method: String, credential: String): Boolean { + fun userId(method: String, credential: String) = + if (method == "appwrite-session") userIdFromSession(credential) else userIdFromJwt(credential) + val user = userId(method, credential) + return user.isNotEmpty() && user == userId(savedMethod, saved) + } + /** * One scheduled run: under a wakelock, reconnect if needed, send a heartbeat when one is due, * re-arm the next run, and call [done] once finished. Shuts down when nothing is saved. + * + * A run of a chained job ([jobId]) arms the next run under the other chained id: scheduling the + * id that is running would stop the run in progress. With [drainMs], the run stays up after + * connecting while the broker replays what was missed (at most that long), so a process woken + * from the background is not cached and frozen before the messages are delivered. */ - fun tick(context: Context, done: () -> Unit) { + fun tick(context: Context, jobId: Int? = null, drainMs: Long = 0, done: () -> Unit) { val app = context.applicationContext val wakeLock = acquireWakeLock(app) worker.execute { @@ -291,12 +428,17 @@ internal object PushBackground { } if (active) { // Arm the next run first, so a run cut short (killed, or a slow broker outlasting - // the wakelock) still leaves the chain armed. A refused credential in converge() - // shuts down, cancelling it again. - schedule(app, INTERVAL_MS) + // the wakelock) still leaves the chain armed; the watchdog job restarts a chain + // that was dropped anyway. A refused credential in converge() shuts down, + // cancelling them again. + schedule(app, INTERVAL_MS, runningJobId = jobId) + ensureWatchdog(app) ensureService(app) registerNetworkCallback(app) converge(app, heartbeat = true) + if (drainMs > 0 && mqtt != null) { + awaitDeliveries(drainMs) + } } else { shutdown(app) } @@ -309,9 +451,49 @@ internal object PushBackground { } } + // Worker thread only. Wait until no message arrived for DRAIN_QUIET_MS and none is being + // delivered, at most [maxMs]. Timed with the real monotonic clock ([nowMs]), which keeps moving + // while this thread sleeps. + private fun awaitDeliveries(maxMs: Long) { + val start = nowMs() + if (lastDelivery < start) { + lastDelivery = start + } + while (nowMs() - start < maxMs) { + if (delivering.get() == 0 && nowMs() - lastDelivery >= DRAIN_QUIET_MS) { + return + } + Thread.sleep(DRAIN_POLL_MS) + } + } + + private fun nowMs(): Long = System.nanoTime() / 1_000_000 + + /** + * Android 12+: hand a run to an expedited job, which starts at once with network access and + * keeps the process runnable until it finishes. A broadcast receiver woken in a cached process + * runs at background priority and can be frozen before it connects. Returns false when the + * job cannot be scheduled (older Android, or the expedited quota is used up), so the caller + * runs the tick itself. + */ + fun runExpedited(context: Context): Boolean { + if (Build.VERSION.SDK_INT < Build.VERSION_CODES.S) { + return false + } + return try { + val job = JobInfo.Builder(EXPEDITED_JOB_ID, ComponentName(context, PushJobService::class.java)) + .setExpedited(true) + .setRequiredNetworkType(JobInfo.NETWORK_TYPE_ANY) + .build() + context.getSystemService(JobScheduler::class.java)?.schedule(job) == JobScheduler.RESULT_SUCCESS + } catch (e: Exception) { + false + } + } + /** Run once more shortly after the task was removed, so an OEM kill that follows is undone. */ fun scheduleRestart(context: Context) { - schedule(context.applicationContext, RESTART_DELAY_MS, jobToo = false) + schedule(context.applicationContext, RESTART_DELAY_MS, alarmOnly = true) } /** @@ -326,6 +508,7 @@ internal object PushBackground { loaded = false } errorHandler = null + connectionHandler = null unregisterNetworkCallback(context.applicationContext) worker.submit { disconnect() }.get() } @@ -361,9 +544,14 @@ internal object PushBackground { // Must hold [lock]. Every background subscription, live or saved, one per filter and title. // Subscriptions sharing a filter and title are saved once, with QoS 1 when any of them wants it. private fun entries(): List = - (saved + listeners.filter { it.background }.flatMap { listener -> listener.topics.map { PushEntry(it, listener.retry, listener.title) } }) + (saved + liveEntries()) .groupBy { it.filter to it.title } - .map { (key, group) -> PushEntry(key.first, group.any { it.retry }, key.second) } + .map { (key, group) -> PushEntry(key.first, group.any { it.retry }, key.second, group.any { it.notifyInForeground }) } + + // Must hold [lock]. + private fun liveEntries(): List = listeners.filter { it.background }.flatMap { listener -> + listener.topics.map { PushEntry(it, listener.retry, listener.title, listener.notifyInForeground) } + } // Must hold [lock]. private fun persist(context: Context) { @@ -397,6 +585,7 @@ internal object PushBackground { private fun activate(context: Context, done: ((Throwable?) -> Unit)? = null) { registerNetworkCallback(context) schedule(context, INTERVAL_MS) + ensureWatchdog(context) ensureService(context) worker.execute { converge(context, heartbeat = false, done) } } @@ -437,6 +626,11 @@ internal object PushBackground { } private fun convergeOnce(context: Context, heartbeat: Boolean) { + // Signed out, or signed in as someone else, since the subscriptions were saved. + if (refreshCredential(context) == CredentialRefresh.GONE) { + stop(context) + return + } val (current, wanted) = synchronized(lock) { config to wanted() } if (current == null || wanted.isEmpty()) { disconnect() @@ -477,24 +671,50 @@ internal object PushBackground { // A failed first connect of a waited-for converge goes to the waiting caller only. val quiet = waiting val connected = AtomicBoolean(false) + val retired = AtomicBoolean(false) val client = buildPushClient( config.copy(keepAlive = KEEP_ALIVE_SECONDS), + onConnected = { + if (!retired.get()) { + isConnected = true + connectionHandler?.invoke(true) + } + }, + onDisconnected = { + if (!retired.get()) { + isConnected = false + connectionHandler?.invoke(false) + } + }, onError = { error -> if (!quiet || connected.get()) { report(error) } }, onAuthRefused = { error -> worker.execute { refused(context, error) } }, + credential = { currentCredential(context, config) }, + retired = retired, ) // Registered before connecting, so the backlog the broker replays right after each // SUBSCRIBE reaches the app however the subscription was made (including the ones the // client restores itself after an automatic reconnect). Acknowledged by hand, once the // message was delivered (see [deliver]). - client.publishes(MqttGlobalPublishFilter.ALL, { publish -> deliver(context, publish) }, true) + client.publishes( + MqttGlobalPublishFilter.ALL, + { publish -> + if (!retired.get()) { + deliver(context, publish) + } + }, + true, + ) try { - client.sendPushConnect(config.copy(keepAlive = KEEP_ALIVE_SECONDS), CONNECT_TIMEOUT_SECONDS) + client.sendPushConnect(config.copy(keepAlive = KEEP_ALIVE_SECONDS), CONNECT_TIMEOUT_SECONDS) { + currentCredential(context, config) + } } catch (e: Exception) { val cause = (e as? ExecutionException)?.cause ?: e + retired.set(true) client.disconnect() val error = connectError(cause) when { @@ -508,6 +728,7 @@ internal object PushBackground { } connected.set(true) mqtt = client + mqttRetired = retired connectedConfig = config connectedNetwork = activeNetwork(context) lastHeartbeat = SystemClock.elapsedRealtime() @@ -575,6 +796,13 @@ internal object PushBackground { // Worker thread only. private fun disconnect() { + mqttRetired?.set(true) + mqttRetired = null + // A retired client's own disconnect is not reported, so report the live one closing here. + if (isConnected) { + isConnected = false + connectionHandler?.invoke(false) + } mqtt?.disconnect() mqtt = null connectedConfig = null @@ -586,6 +814,21 @@ internal object PushBackground { // registered the message is also saved, and the next onError registration receives it. A // waiting caller receives it instead. private fun refused(context: Context, error: Throwable) { + // The saved credential may be a session the app has since rotated: retry once with the + // current one. Refused again, it is unchanged and delivery stops below. + if (!waiting) { + when (refreshCredential(context)) { + CredentialRefresh.CHANGED -> { + converge(context, heartbeat = false) + return + } + CredentialRefresh.GONE -> { + stop(context) + return + } + CredentialRefresh.UNCHANGED -> Unit + } + } if (waiting) { fail(error) } else if (!report(error)) { @@ -596,11 +839,24 @@ internal object PushBackground { // Deliver a message, then acknowledge it: right away, or once every listener that acknowledges // messages itself has done so (at most ACK_TIMEOUT_SECONDS later). + // A message counts as being delivered until it is acknowledged: for the React Native and Flutter + // bridges, once their callback has run (or the acknowledgement times out), not when the event is + // handed over, so a run's drain window waits for it. private fun deliver(context: Context, publish: Mqtt5Publish) { + delivering.incrementAndGet() + lastDelivery = nowMs() + deliverAcknowledged(context, publish) { + lastDelivery = nowMs() + delivering.decrementAndGet() + } + } + + private fun deliverAcknowledged(context: Context, publish: Mqtt5Publish, settled: () -> Unit) { val acknowledged = AtomicBoolean(false) val acknowledge = { if (acknowledged.compareAndSet(false, true)) { runCatching { publish.acknowledge() } + settled() } } val pending = AtomicInteger(1) @@ -624,7 +880,7 @@ internal object PushBackground { settle() } if (!acknowledged.get()) { - worker.schedule({ acknowledge() }, ACK_TIMEOUT_SECONDS, TimeUnit.SECONDS) + ackTimer.schedule({ acknowledge() }, ACK_TIMEOUT_SECONDS, TimeUnit.SECONDS) } } @@ -656,8 +912,13 @@ internal object PushBackground { // The app's PushReceiver gets only messages no live callback received. val handled = matching.isEmpty() && deliverToReceivers(context, message) val content = notificationContent(message) - val titles = matching.filter { it.background }.map { it.title ?: message.topic } + - if (handled) emptyList() else entries.map { it.title ?: message.topic } + // Background subscriptions notify while the app is not visible; on screen, only those + // that opted in with notifyInForeground do, since the app shows the message itself. A + // message no live callback received still notifies, as the app has not shown it. + val foreground = appInForeground() + val shown = foreground && matching.isNotEmpty() + val titles = matching.filter { it.background && (!foreground || it.notifyInForeground) }.map { it.title ?: message.topic } + + if (handled) emptyList() else entries.filter { !shown || it.notifyInForeground }.map { it.title ?: message.topic } // A title the server sent replaces every subscription's, so one notification is posted. titles.map { content.title ?: it }.distinct().forEach { notify(context, message, it, content) } } finally { @@ -712,14 +973,13 @@ internal object PushBackground { builder.setLargeIcon(image) .setStyle(NotificationCompat.BigPictureStyle().bigPicture(image).bigLargeIcon(null as Bitmap?).setSummaryText(body)) } - context.packageManager.getLaunchIntentForPackage(context.packageName)?.let { launch -> - launch.addFlags(Intent.FLAG_ACTIVITY_NEW_TASK or Intent.FLAG_ACTIVITY_SINGLE_TOP) - .putExtra(EXTRA_TOPIC, message.topic) - .putExtra(EXTRA_PAYLOAD, message.data) - builder.setContentIntent( - PendingIntent.getActivity(context, id, launch, PendingIntent.FLAG_UPDATE_CURRENT or PendingIntent.FLAG_IMMUTABLE), - ) - } + val open = Intent(context, PushOpenActivity::class.java) + .addFlags(Intent.FLAG_ACTIVITY_NEW_TASK or Intent.FLAG_ACTIVITY_NO_ANIMATION) + .putExtra(EXTRA_TOPIC, message.topic) + .putExtra(EXTRA_PAYLOAD, message.data) + builder.setContentIntent( + PendingIntent.getActivity(context, id, open, PendingIntent.FLAG_UPDATE_CURRENT or PendingIntent.FLAG_IMMUTABLE), + ) try { manager.notify(id, builder.build()) } catch (e: SecurityException) { @@ -727,6 +987,11 @@ internal object PushBackground { } } + // Whether one of the app's activities is visible: a foreground service alone does not count. + private fun appInForeground(): Boolean = ActivityManager.RunningAppProcessInfo() + .also { ActivityManager.getMyMemoryState(it) } + .importance == ActivityManager.RunningAppProcessInfo.IMPORTANCE_FOREGROUND + /** The server's `notification` block in [message]: nulls when the payload has none or is not JSON. */ fun notificationContent(message: PushMessage): PushNotificationContent { val notification = runCatching { JSONObject(message.data) }.getOrNull()?.optJSONObject("notification") @@ -736,6 +1001,11 @@ internal object PushBackground { } // Downloads notification images, so a slow one is abandoned without holding up delivery. + // Acknowledgement timeouts, apart from [worker] so they fire while a run's drain window holds it. + private val ackTimer = Executors.newSingleThreadScheduledExecutor { runnable -> + Thread(runnable, "AppwritePushAck").apply { isDaemon = true } + } + private val imageLoader = Executors.newCachedThreadPool { runnable -> Thread(runnable, "AppwritePushImage").apply { isDaemon = true } } @@ -746,7 +1016,7 @@ internal object PushBackground { val connection = runCatching { URL(url).openConnection() as HttpURLConnection }.getOrNull() ?: return null connection.connectTimeout = IMAGE_TIMEOUT_MS connection.readTimeout = IMAGE_TIMEOUT_MS - val download = imageLoader.submit { connection.inputStream.use { BitmapFactory.decodeStream(it) } } + val download = imageLoader.submit { connection.inputStream.use { readLimited(it, IMAGE_MAX_BYTES) }?.let { decodeImage(it) } } return try { download.get(IMAGE_TIMEOUT_MS.toLong(), TimeUnit.MILLISECONDS) } catch (e: Exception) { @@ -757,6 +1027,37 @@ internal object PushBackground { } } + // At most [limit] bytes of [input], or null when it holds more. + private fun readLimited(input: InputStream, limit: Int): ByteArray? { + val out = ByteArrayOutputStream() + val buffer = ByteArray(8_192) + while (true) { + val read = input.read(buffer) + if (read < 0) { + return out.toByteArray() + } + if (out.size() + read > limit) { + return null + } + out.write(buffer, 0, read) + } + } + + // Decode [bytes] downsampled so neither side exceeds IMAGE_MAX_PX, so a large photo cannot + // exhaust memory while a notification is posted. + private fun decodeImage(bytes: ByteArray): Bitmap? { + val bounds = BitmapFactory.Options().apply { inJustDecodeBounds = true } + BitmapFactory.decodeByteArray(bytes, 0, bytes.size, bounds) + if (bounds.outWidth <= 0 || bounds.outHeight <= 0) { + return null + } + var sample = 1 + while (bounds.outWidth / sample > IMAGE_MAX_PX || bounds.outHeight / sample > IMAGE_MAX_PX) { + sample *= 2 + } + return BitmapFactory.decodeByteArray(bytes, 0, bytes.size, BitmapFactory.Options().apply { inSampleSize = sample }) + } + /** The ongoing notification the foreground service shows, on its own quiet channel. */ fun ongoingNotification(context: Context) = NotificationCompat.Builder(context, SERVICE_CHANNEL_ID) .also { createChannels(context) } @@ -815,19 +1116,16 @@ internal object PushBackground { } // Arm the next run in [delayMs]: the job, and an alarm just after it in case the job is late. - private fun schedule(context: Context, delayMs: Long, jobToo: Boolean = true) { - val interval = if (delayMs == INTERVAL_MS && privileged(context)) PRIVILEGED_INTERVAL_MS else delayMs - if (jobToo) { - try { - val job = JobInfo.Builder(JOB_ID, ComponentName(context, PushJobService::class.java)) - .setMinimumLatency(interval) - .setRequiredNetworkType(JobInfo.NETWORK_TYPE_ANY) - .setPersisted(true) - .build() - context.getSystemService(JobScheduler::class.java)?.schedule(job) - } catch (e: Exception) { - report(e) + // The chained job alternates between two ids, so a run ([runningJobId]) never schedules its + // own id, which would stop it; the other id is cancelled only when it is not the one running. + private fun schedule(context: Context, delayMs: Long, alarmOnly: Boolean = false, runningJobId: Int? = null) { + val interval = interval(context, delayMs) + if (!alarmOnly) { + val next = if (runningJobId == JOB_ID) NEXT_JOB_ID else JOB_ID + if (runningJobId == null) { + context.getSystemService(JobScheduler::class.java)?.cancel(if (next == JOB_ID) NEXT_JOB_ID else JOB_ID) } + scheduleJob(context, delayMs, next) } val alarms = context.getSystemService(AlarmManager::class.java) ?: return val at = SystemClock.elapsedRealtime() + interval + 1_000L @@ -843,8 +1141,48 @@ internal object PushBackground { } } + private fun interval(context: Context, delayMs: Long): Long = + if (delayMs == INTERVAL_MS && privileged(context)) PRIVILEGED_INTERVAL_MS else delayMs + + private fun scheduleJob(context: Context, delayMs: Long, id: Int) { + try { + val job = JobInfo.Builder(id, ComponentName(context, PushJobService::class.java)) + .setMinimumLatency(interval(context, delayMs)) + .setRequiredNetworkType(JobInfo.NETWORK_TYPE_ANY) + .setPersisted(true) + .build() + context.getSystemService(JobScheduler::class.java)?.schedule(job) + } catch (e: Exception) { + report(e) + } + } + + // A periodic job, independent of the chain of runs, that restarts the chain when a run was + // dropped (the process killed between runs, a deferred alarm, a throttled job). Scheduled once, + // so its period is not reset by every run. + private fun ensureWatchdog(context: Context) { + val jobs = context.getSystemService(JobScheduler::class.java) ?: return + try { + if (jobs.allPendingJobs.any { it.id == WATCHDOG_JOB_ID }) { + return + } + jobs.schedule( + JobInfo.Builder(WATCHDOG_JOB_ID, ComponentName(context, PushJobService::class.java)) + .setPeriodic(WATCHDOG_INTERVAL_MS) + .setRequiredNetworkType(JobInfo.NETWORK_TYPE_ANY) + .setPersisted(true) + .build(), + ) + } catch (e: Exception) { + report(e) + } + } + private fun cancelSchedule(context: Context) { context.getSystemService(JobScheduler::class.java)?.cancel(JOB_ID) + context.getSystemService(JobScheduler::class.java)?.cancel(NEXT_JOB_ID) + context.getSystemService(JobScheduler::class.java)?.cancel(EXPEDITED_JOB_ID) + context.getSystemService(JobScheduler::class.java)?.cancel(WATCHDOG_JOB_ID) context.getSystemService(AlarmManager::class.java)?.cancel(tickIntent(context)) } @@ -855,6 +1193,63 @@ internal object PushBackground { PendingIntent.FLAG_UPDATE_CURRENT or PendingIntent.FLAG_IMMUTABLE, ) + /** What background delivery can rely on now; see [PushBackgroundStatus]. */ + fun backgroundStatus(context: Context): PushBackgroundStatus { + val app = context.applicationContext + return PushBackgroundStatus( + exactAlarms = Build.VERSION.SDK_INT < Build.VERSION_CODES.S || + app.getSystemService(AlarmManager::class.java)?.canScheduleExactAlarms() == true, + ignoringBatteryOptimizations = app.getSystemService(PowerManager::class.java)?.isIgnoringBatteryOptimizations(app.packageName) == true, + foregroundService = PushStore.foreground(app), + ) + } + + /** + * Open the system screen where the user allows exact alarms (Android 12+). Returns false when + * there is nothing to ask: already allowed, older Android, or the app does not declare + * `SCHEDULE_EXACT_ALARM`. + */ + fun requestExactAlarms(context: Context): Boolean { + if (Build.VERSION.SDK_INT < Build.VERSION_CODES.S || backgroundStatus(context).exactAlarms) { + return false + } + return openSettings(context, Settings.ACTION_REQUEST_SCHEDULE_EXACT_ALARM, Manifest.permission.SCHEDULE_EXACT_ALARM) + } + + /** + * Ask the user to exempt the app from battery optimisation. Returns false when there is + * nothing to ask: already exempt, or the app does not declare + * `REQUEST_IGNORE_BATTERY_OPTIMIZATIONS`. + */ + fun requestIgnoreBatteryOptimizations(context: Context): Boolean { + if (backgroundStatus(context).ignoringBatteryOptimizations) { + return false + } + return openSettings( + context, + Settings.ACTION_REQUEST_IGNORE_BATTERY_OPTIMIZATIONS, + Manifest.permission.REQUEST_IGNORE_BATTERY_OPTIMIZATIONS, + ) + } + + // Open a settings screen for this package when the app declares [permission]. + private fun openSettings(context: Context, action: String, permission: String): Boolean { + val app = context.applicationContext + val declared = runCatching { + @Suppress("DEPRECATION") + app.packageManager.getPackageInfo(app.packageName, PackageManager.GET_PERMISSIONS).requestedPermissions + }.getOrNull()?.contains(permission) == true + if (!declared) { + return false + } + return try { + app.startActivity(Intent(action, Uri.parse("package:${app.packageName}")).addFlags(Intent.FLAG_ACTIVITY_NEW_TASK)) + true + } catch (e: Exception) { + false + } + } + // Allowed exact alarms or exempt from battery optimisation: the alarm is reliable, so the // runs can be spaced further apart. private fun privileged(context: Context): Boolean = @@ -945,6 +1340,7 @@ internal object PushStore { authMethod = it.getString("authMethod"), credential = it.getString("credential"), project = it.getString("project"), + sessionCookieUrl = if (it.isNull("sessionCookieUrl")) null else it.optString("sessionCookieUrl").ifEmpty { null }, ) } val list = json.getJSONArray("entries") @@ -954,6 +1350,7 @@ internal object PushStore { filter = entry.getString("filter"), retry = entry.getBoolean("retry"), title = if (entry.isNull("title")) null else entry.getString("title"), + notifyInForeground = entry.optBoolean("notifyInForeground", false), ) } config to entries @@ -973,11 +1370,12 @@ internal object PushStore { .put("keepAlive", config.keepAlive) .put("authMethod", config.authMethod) .put("credential", config.credential) - .put("project", config.project), + .put("project", config.project) + .put("sessionCookieUrl", config.sessionCookieUrl ?: JSONObject.NULL), ) .put( "entries", - JSONArray(entries.map { JSONObject().put("filter", it.filter).put("retry", it.retry).put("title", it.title ?: JSONObject.NULL) }), + JSONArray(entries.map { JSONObject().put("filter", it.filter).put("retry", it.retry).put("title", it.title ?: JSONObject.NULL).put("notifyInForeground", it.notifyInForeground) }), ) write(context, STATE_FILE, json) } diff --git a/android/src/main/kotlin/io/appwrite/services/PushBridge.kt b/android/src/main/kotlin/io/appwrite/services/PushBridge.kt index d7b79b5d..e5cb5aae 100644 --- a/android/src/main/kotlin/io/appwrite/services/PushBridge.kt +++ b/android/src/main/kotlin/io/appwrite/services/PushBridge.kt @@ -24,6 +24,8 @@ class PushBridge(context: Context, private val events: Events) { fun onMessage(subscriptionId: String, message: PushMessage, ackToken: String) fun onError(message: String) + + fun onConnection(connected: Boolean) } private val appContext = context.applicationContext @@ -37,13 +39,17 @@ class PushBridge(context: Context, private val events: Events) { @Volatile private var errorCallback = false + private var connected: Boolean? = null + /** * Host the subscriptions in [subscriptionsJson] on the background connection described by * [configJson]. * * Config: `{"host", "port", "tls", "tlsInsecure", "clientId" (optional; per user and install - * when empty), "authMethod", "credential", "project"}`. Subscriptions: an array of - * `{"id", "topic", "background", "title" (optional), "retry"}`. + * when empty), "authMethod", "credential", "project", "sessionCookieUrl" (optional: the endpoint + * whose session cookie in the WebView cookie store holds the credential, so background runs + * follow a rotated session)}`. Subscriptions: an array of + * `{"id", "topic", "background", "title" (optional), "retry", "notifyInForeground" (optional)}`. * * [done] is called once the connection is up and every filter is subscribed, or with the * failure's message; the SDK rejects its subscribe with it and reports it itself. @@ -62,6 +68,7 @@ class PushBridge(context: Context, private val events: Events) { authMethod = authMethod, credential = credential, project = json.optString("project"), + sessionCookieUrl = if (json.isNull("sessionCookieUrl")) null else json.optString("sessionCookieUrl").ifEmpty { null }, ) val list = JSONArray(subscriptionsJson) val listeners = (0 until list.length()).map { index -> @@ -80,6 +87,7 @@ class PushBridge(context: Context, private val events: Events) { background = subscription.optBoolean("background", false), title = if (subscription.isNull("title")) null else subscription.optString("title").ifEmpty { null }, retry = subscription.optBoolean("retry", true), + notifyInForeground = subscription.optBoolean("notifyInForeground", false), ) } PushBackground.errorHandler = { error -> @@ -88,21 +96,54 @@ class PushBridge(context: Context, private val events: Events) { } errorCallback } + PushBackground.connectionHandler = { connection(it) } PushBackground.host(appContext, config, listeners) { error -> + if (error == null && PushBackground.isConnected) { + connection(true) + } done(error?.let { it.message ?: it.toString() }) } } + /** Whether the background connection is open now. */ + fun isConnected(): Boolean = PushBackground.isConnected + + private fun connection(open: Boolean) { + synchronized(this) { + if (connected == open) { + return + } + connected = open + } + events.onConnection(open) + } + /** Acknowledge a message once the SDK's callback for it has run. */ fun ack(token: String) { acks.remove(token)?.second?.invoke() } /** Stop hosting the live subscriptions, keeping the ones saved by an earlier run. */ - fun release() = PushBackground.release(appContext) + fun release() { + forgetConnection() + PushBackground.release(appContext) + } /** Stop background delivery and forget every saved subscription (sign-out). */ - fun stop() = PushBackground.stop(appContext) + fun stop() { + forgetConnection() + PushBackground.stop(appContext) + } + + private fun forgetConnection() { + PushBackground.connectionHandler = null + val wasOpen = synchronized(this) { + (connected == true).also { connected = null } + } + if (wasOpen) { + events.onConnection(false) + } + } /** Run background delivery in a foreground service; saved across restarts. */ fun setForeground(enabled: Boolean) = PushBackground.setForeground(appContext, enabled) @@ -110,8 +151,14 @@ class PushBridge(context: Context, private val events: Events) { /** Whether an earlier run saved background subscriptions that are still delivered. */ fun hasSaved(): Boolean = PushBackground.hasSaved(appContext) - /** Resume saved background delivery now instead of at the next scheduled run. */ - fun resume() = PushBackground.resume(appContext) + /** + * Resume saved background delivery now instead of at the next scheduled run, with the app's + * current credential (null when it has none). A rotated session of the same user replaces + * the saved one; another user, or none when [signedOutWhenMissing], drops the saved + * subscriptions without reporting an error. + */ + fun resume(authMethod: String?, credential: String?, signedOutWhenMissing: Boolean) = + PushBackground.resume(appContext, authMethod, credential, signedOutWhenMissing) /** * Record whether the app has an onError callback. Returns the refusal that stopped background @@ -122,6 +169,22 @@ class PushBridge(context: Context, private val events: Events) { return if (registered) PushStore.takeStoppedError(appContext) else null } + /** What background delivery can rely on, as `{"exactAlarms", "ignoringBatteryOptimizations", "foregroundService", "bestEffort"}`. */ + fun backgroundStatus(): String = PushBackground.backgroundStatus(appContext).let { + JSONObject() + .put("exactAlarms", it.exactAlarms) + .put("ignoringBatteryOptimizations", it.ignoringBatteryOptimizations) + .put("foregroundService", it.foregroundService) + .put("bestEffort", it.bestEffort) + .toString() + } + + /** Open the system screen that allows exact alarms; false when there is nothing to ask. */ + fun requestExactAlarms(): Boolean = PushBackground.requestExactAlarms(appContext) + + /** Ask to exempt the app from battery optimisation; false when there is nothing to ask. */ + fun requestIgnoreBatteryOptimizations(): Boolean = PushBackground.requestIgnoreBatteryOptimizations(appContext) + /** The default client id for a credential, so a foreground connection shares its session. */ fun defaultClientId(authMethod: String, credential: String): String = defaultPushClientId(appContext, authMethod, credential) diff --git a/android/src/main/kotlin/io/appwrite/services/PushCore.kt b/android/src/main/kotlin/io/appwrite/services/PushCore.kt index 20c56242..15f86b6a 100644 --- a/android/src/main/kotlin/io/appwrite/services/PushCore.kt +++ b/android/src/main/kotlin/io/appwrite/services/PushCore.kt @@ -12,6 +12,7 @@ import com.hivemq.client.mqtt.mqtt5.Mqtt5ClientConfig import com.hivemq.client.mqtt.mqtt5.auth.Mqtt5EnhancedAuthMechanism import com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException import com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5DisconnectException +import com.hivemq.client.mqtt.mqtt5.lifecycle.Mqtt5ClientReconnector import com.hivemq.client.mqtt.mqtt5.message.auth.Mqtt5Auth import com.hivemq.client.mqtt.mqtt5.message.auth.Mqtt5AuthBuilder import com.hivemq.client.mqtt.mqtt5.message.auth.Mqtt5EnhancedAuthBuilder @@ -30,6 +31,7 @@ import java.util.concurrent.CompletableFuture import java.util.concurrent.ExecutionException import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicReference import javax.net.ssl.ManagerFactoryParameters import javax.net.ssl.TrustManager import javax.net.ssl.TrustManagerFactory @@ -49,6 +51,24 @@ data class PushMessage( get() = String(payload, Charsets.UTF_8) } +/** + * What background delivery can rely on. Without exact alarms, the scheduled wake-ups are inexact + * and Doze can defer them, so delivery is [bestEffort] unless foreground mode keeps the + * connection open. + */ +data class PushBackgroundStatus( + /** The app may schedule exact alarms (`SCHEDULE_EXACT_ALARM`, granted). */ + val exactAlarms: Boolean, + /** The app is exempt from battery optimisation. */ + val ignoringBatteryOptimizations: Boolean, + /** Foreground mode (`setForeground(true)`) keeps the connection open in a service. */ + val foregroundService: Boolean, +) { + /** Wake-ups may be deferred by Doze, so messages can arrive late while the app is closed. */ + val bestEffort: Boolean + get() = !exactAlarms && !foregroundService +} + /** Connection parameters shared between the in-process client and the foreground service. */ internal data class PushConfig( val host: String, @@ -60,6 +80,10 @@ internal data class PushConfig( val authMethod: String, val credential: String, val project: String, + // The endpoint whose `a_session_` cookie in the WebView cookie store holds the app's + // session, when its credential comes from there (React Native). Background runs read it, so + // they keep up with a rotated session while no app code runs. + val sessionCookieUrl: String? = null, ) /** @@ -110,6 +134,8 @@ internal fun buildPushClient( onDisconnected: (() -> Unit)? = null, onError: ((Throwable) -> Unit)? = null, onAuthRefused: ((Throwable) -> Unit)? = null, + credential: () -> String? = { config.credential }, + retired: AtomicBoolean? = null, ): Mqtt5AsyncClient { // config.clientId is stable per user (see buildConfig), so a reconnect rebuild reuses it // and resumes the same session. @@ -130,11 +156,25 @@ internal fun buildPushClient( // Whether this client has connected yet: until it has, a refused CONNECT is final. val connected = AtomicBoolean(false) + val client = AtomicReference() builder = builder.addConnectedListener { + // A reconnect that was already under way when the client was retired disconnects at once, + // so it never competes with its replacement for the broker session. + if (retired?.get() == true) { + client.get()?.disconnect() + return@addConnectedListener + } connected.set(true) onConnected?.invoke() } builder = builder.addDisconnectedListener { context -> + // A client its owner replaced or closed ([retired]) never reconnects: one still retrying + // in the background would otherwise come back as a second connection for the same id. + if (retired?.get() == true) { + context.reconnector.reconnect(false) + onDisconnected?.invoke() + return@addDisconnectedListener + } // A refused first CONNECT (bad credential, rate limit) is final: stop the automatic // reconnect so pushConnect fails with the broker's reason instead of retrying forever. val refused = !connected.get() && context.cause is Mqtt5ConnAckException @@ -142,8 +182,16 @@ internal fun buildPushClient( // re-auth) is final too, instead of being retried with the same credential forever. val authRefused = onAuthRefused != null && connected.get() && context.source != MqttDisconnectSource.USER && isAuthRefusal(context.cause) - if (refused || authRefused) { + // Authentication was aborted because the app has no current credential (signed out): there + // is nothing to reconnect with, so the attempt fails now instead of retrying until timeout. + val noCredential = generateSequence(context.cause) { it.cause }.any { it is PushNoCredentialException } + if (refused || authRefused || noCredential) { context.reconnector.reconnect(false) + } else if (context.reconnector.isReconnect) { + // HiveMQ's automatic reconnect sends a CONNECT it builds itself, without user + // properties: the broker then misses the project and refuses the credential. Reconnect + // with the same CONNECT this client connects with instead. + (context.reconnector as? Mqtt5ClientReconnector)?.connect(pushConnectMessage(config, credential)) } onDisconnected?.invoke() if (!refused) { @@ -168,7 +216,7 @@ internal fun buildPushClient( } } - return builder.buildAsync() + return builder.buildAsync().also { client.set(it) } } /** @@ -207,10 +255,31 @@ internal fun isAuthRefusal(error: Throwable?): Boolean = when (error) { else -> false } -/** Send the CONNECT and wait for the CONNACK, at most [timeoutSeconds] when given. */ -internal fun Mqtt5AsyncClient.sendPushConnect(config: PushConfig, timeoutSeconds: Long? = null) { - val connAck = connectWith() - .enhancedAuth(PushAuthMechanism(config.authMethod, config.credential)) +/** + * Send the CONNECT and wait for the CONNACK, at most [timeoutSeconds] when given. [credential] + * supplies the credential each time one is sent, including on the client's automatic reconnects + * and re-auths, so they can use a session rotated since this CONNECT; null aborts the attempt. + */ +internal fun Mqtt5AsyncClient.sendPushConnect( + config: PushConfig, + timeoutSeconds: Long? = null, + credential: () -> String? = { config.credential }, +) { + val connAck = connect(pushConnectMessage(config, credential)) + if (timeoutSeconds == null) { + connAck.get() + } else { + connAck.get(timeoutSeconds, TimeUnit.SECONDS) + } +} + +/** + * The CONNECT for [config]: enhanced auth with the credential from [credential], and the project + * as a user property, which the broker resolves the credential against. + */ +internal fun pushConnectMessage(config: PushConfig, credential: () -> String?): Mqtt5Connect = + Mqtt5Connect.builder() + .enhancedAuth(PushAuthMechanism(config.authMethod, config.project, credential)) // Clean start is always off, so the broker keeps this client's session and can // redeliver missed messages to QoS-1 subscriptions on reconnect. .cleanStart(false) @@ -218,13 +287,7 @@ internal fun Mqtt5AsyncClient.sendPushConnect(config: PushConfig, timeoutSeconds .userProperties() .add("projectId", config.project) .applyUserProperties() - .send() - if (timeoutSeconds == null) { - connAck.get() - } else { - connAck.get(timeoutSeconds, TimeUnit.SECONDS) - } -} + .build() /** * The error a lost connection reports, or null when it is not an error (a disconnect the app @@ -266,11 +329,14 @@ internal fun Mqtt5Publish.toPushMessage(): PushMessage = /** * Single-step MQTT 5 enhanced auth: the method and credential are placed in the CONNECT - * packet and the broker accepts them in the CONNACK, with no AUTH round-trip. + * packet and the broker accepts them in the CONNACK, with no AUTH round-trip. The credential is + * read from [credential] each time, so an automatic reconnect or re-auth sends the current one; + * when it has none, the attempt fails instead of sending a credential known to be stale. */ internal class PushAuthMechanism( private val method: String, - private val credential: String, + private val project: String, + private val credential: () -> String?, ) : Mqtt5EnhancedAuthMechanism { override fun getMethod(): MqttUtf8String = MqttUtf8String.of(method) @@ -282,7 +348,8 @@ internal class PushAuthMechanism( connect: Mqtt5Connect, authBuilder: Mqtt5EnhancedAuthBuilder, ): CompletableFuture { - authBuilder.data(credential.toByteArray(Charsets.UTF_8)) + val current = credential() ?: return failedAuth() + authBuilder.data(current.toByteArray(Charsets.UTF_8)) return CompletableFuture.completedFuture(null) } @@ -290,7 +357,12 @@ internal class PushAuthMechanism( clientConfig: Mqtt5ClientConfig, authBuilder: Mqtt5AuthBuilder, ): CompletableFuture { - authBuilder.data(credential.toByteArray(Charsets.UTF_8)) + val current = credential() ?: return failedAuth() + // The broker resolves a re-auth against the project in its user properties too. + authBuilder.data(current.toByteArray(Charsets.UTF_8)) + .userProperties() + .add("projectId", project) + .applyUserProperties() return CompletableFuture.completedFuture(null) } @@ -317,8 +389,14 @@ internal class PushAuthMechanism( override fun onAuthError(clientConfig: Mqtt5ClientConfig, cause: Throwable) = Unit override fun onReAuthError(clientConfig: Mqtt5ClientConfig, cause: Throwable) = Unit + + private fun failedAuth(): CompletableFuture = + CompletableFuture().apply { completeExceptionally(PushNoCredentialException()) } } +/** Authentication aborted because there is no current credential to send (the app signed out). */ +internal class PushNoCredentialException : IllegalStateException("No current credential to authenticate with") + /** * A [TrustManagerFactory] that accepts any certificate — used when `tlsInsecure` is set * (for a broker whose certificate does not chain to a public root). @@ -353,6 +431,8 @@ internal class PushListener( // Called instead of [callback] when set, with the acknowledgement to run once the message was // handled: the React Native and Flutter bridges acknowledge after their callback has run. val acknowledgingCallback: ((PushMessage, () -> Unit) -> Unit)? = null, + // Also post a background notification while the app is visible. + val notifyInForeground: Boolean = false, ) /** MQTT topic-filter match with '+' (single level) and '#' (multi level). */ diff --git a/android/src/main/kotlin/io/appwrite/services/PushOpenActivity.kt b/android/src/main/kotlin/io/appwrite/services/PushOpenActivity.kt new file mode 100644 index 00000000..354ca6f0 --- /dev/null +++ b/android/src/main/kotlin/io/appwrite/services/PushOpenActivity.kt @@ -0,0 +1,57 @@ +package io.appwrite.services + +import android.app.Activity +import android.content.Intent +import android.os.Bundle + +class PushOpenActivity : Activity() { + override fun onCreate(savedInstanceState: Bundle?) { + super.onCreate(savedInstanceState) + val topic = intent.getStringExtra(PushBackground.EXTRA_TOPIC) + val payload = intent.getStringExtra(PushBackground.EXTRA_PAYLOAD) + packageManager.getLaunchIntentForPackage(packageName)?.let { launch -> + launch.addFlags(Intent.FLAG_ACTIVITY_NEW_TASK or Intent.FLAG_ACTIVITY_SINGLE_TOP) + if (topic != null && payload != null) { + launch.putExtra(PushBackground.EXTRA_TOPIC, topic).putExtra(PushBackground.EXTRA_PAYLOAD, payload) + } + startActivity(launch) + } + // After opening the app, so a screen a tap callback opens lands on top of it. + if (topic != null && payload != null) { + PushTaps.record(PushTap(topic, payload)) + } + finish() + } +} + +internal data class PushTap(val topic: String, val payload: String) + +internal object PushTaps { + private var pending: PushTap? = null + private val listeners = mutableListOf<(PushTap) -> Unit>() + + fun record(tap: PushTap) { + val deliver = synchronized(this) { + listeners.toList().also { + if (it.isEmpty()) { + pending = tap + } + } + } + deliver.forEach { it(tap) } + } + + @Synchronized + fun take(): PushTap? = pending.also { pending = null } + + fun listen(onTap: (PushTap) -> Unit): () -> Unit { + synchronized(this) { + listeners.add(onTap) + } + return { + synchronized(this) { + listeners.remove(onTap) + } + } + } +} diff --git a/android/src/main/kotlin/io/appwrite/services/PushWakeups.kt b/android/src/main/kotlin/io/appwrite/services/PushWakeups.kt index 091fd528..c6d56165 100644 --- a/android/src/main/kotlin/io/appwrite/services/PushWakeups.kt +++ b/android/src/main/kotlin/io/appwrite/services/PushWakeups.kt @@ -12,11 +12,15 @@ import android.content.Intent */ class PushJobService : JobService() { override fun onStartJob(params: JobParameters): Boolean { - PushBackground.tick(this) { jobFinished(params, false) } + // A chained job arms the next under the other chained id; the watchdog and an expedited + // run (handed over by the alarm) only run once. Each stays up while missed messages arrive. + val chained = params.jobId.takeIf { it == PushBackground.JOB_ID || it == PushBackground.NEXT_JOB_ID } + PushBackground.tick(this, chained, PushBackground.JOB_DRAIN_MS) { jobFinished(params, false) } return true } - // The run re-arms the next one itself, so a stopped run needs no retry. + // The run re-arms the next one itself, and the watchdog repeats on its own, so a stopped run + // needs no retry. override fun onStopJob(params: JobParameters): Boolean = false } @@ -29,8 +33,13 @@ class PushAlarmReceiver : BroadcastReceiver() { if (intent.action != PushBackground.ACTION_TICK) { return } + // On Android 12+ the run goes to an expedited job, which keeps the process runnable while it + // connects and the broker replays; it runs here when the system refuses one. + if (PushBackground.runExpedited(context)) { + return + } val pending = goAsync() - PushBackground.tick(context) { pending.finish() } + PushBackground.tick(context, drainMs = PushBackground.RECEIVER_DRAIN_MS) { pending.finish() } } } diff --git a/lib/src/client_browser.dart b/lib/src/client_browser.dart index 4caa7081..e8f4ada5 100644 --- a/lib/src/client_browser.dart +++ b/lib/src/client_browser.dart @@ -40,7 +40,7 @@ class ClientBrowser extends ClientBase with ClientMixin { 'x-sdk-name': 'Flutter', 'x-sdk-platform': 'client', 'x-sdk-language': 'flutter', - 'x-sdk-version': '27.1.0-rc.3', + 'x-sdk-version': '27.1.0-rc.7', 'X-Appwrite-Response-Format': '2.3.0', }; diff --git a/lib/src/client_io.dart b/lib/src/client_io.dart index 5b4dcde9..ebfc79c1 100644 --- a/lib/src/client_io.dart +++ b/lib/src/client_io.dart @@ -60,7 +60,7 @@ class ClientIO extends ClientBase with ClientMixin { 'x-sdk-name': 'Flutter', 'x-sdk-platform': 'client', 'x-sdk-language': 'flutter', - 'x-sdk-version': '27.1.0-rc.3', + 'x-sdk-version': '27.1.0-rc.7', 'X-Appwrite-Response-Format': '2.3.0', }; diff --git a/lib/src/mqtt.dart b/lib/src/mqtt.dart index d55f1e8e..1595ff0b 100644 --- a/lib/src/mqtt.dart +++ b/lib/src/mqtt.dart @@ -1,3 +1,5 @@ +import 'dart:convert'; + import 'mqtt_browser.dart' if (dart.library.io) 'mqtt_io.dart'; import 'service.dart'; import 'client.dart'; @@ -16,11 +18,72 @@ abstract class PushSubscription { /// Drop this subscription. Closes the connection once the last one is gone. void unsubscribe(); - /// Live-update this subscription's [background] delivery, notification [title] and/or - /// [retry] (QoS). Only the arguments you pass change; the rest stay as they - /// were. Changing [retry] re-subscribes this subscription's topics at the - /// new QoS. - void update({bool? background, String? title, bool? retry}); + /// Live-update this subscription's [background] delivery, notification [title], [retry] + /// (QoS) and/or [notifyInForeground]. Only the arguments you pass change; the rest stay as + /// they were. Changing [retry] re-subscribes this subscription's topics at the new QoS. + void update({ + bool? background, + String? title, + bool? retry, + bool? notifyInForeground, + }); +} + +/// What Android background delivery can rely on; see [Push.backgroundStatus]. +class PushBackgroundStatus { + /// The app may schedule exact alarms (`SCHEDULE_EXACT_ALARM`, granted). + final bool exactAlarms; + + /// The app is exempt from battery optimisation. + final bool ignoringBatteryOptimizations; + + /// Foreground mode ([Push.setForeground]) keeps the connection open in a service. + final bool foregroundService; + + /// Wake-ups may be deferred by Doze, so messages can arrive late while the app is closed. + final bool bestEffort; + + const PushBackgroundStatus({ + required this.exactAlarms, + required this.ignoringBatteryOptimizations, + required this.foregroundService, + required this.bestEffort, + }); + + factory PushBackgroundStatus.fromMap(Map map) => + PushBackgroundStatus( + exactAlarms: map['exactAlarms'] == true, + ignoringBatteryOptimizations: + map['ignoringBatteryOptimizations'] == true, + foregroundService: map['foregroundService'] == true, + bestEffort: map['bestEffort'] == true, + ); +} + +/// A background notification the user tapped; see [Push.getInitialNotification] and +/// [Push.onNotificationOpened]. +class PushNotificationOpened { + /// The topic the message was published to. + final String topic; + + /// The `data` sent with the message (e.g. `createPush`), parsed; empty when it had none. + final Map data; + + const PushNotificationOpened({required this.topic, required this.data}); + + /// The tap on a notification for a message on [topic] with the raw [payload]. + factory PushNotificationOpened.fromPayload(String topic, String payload) { + Object? data; + try { + data = (jsonDecode(payload) as Map?)?['data']; + } catch (_) { + data = null; + } + return PushNotificationOpened( + topic: topic, + data: data is Map ? data : {}, + ); + } } /// Appwrite native push service — the realtime analog delivered over an MQTT @@ -34,9 +97,9 @@ abstract class PushSubscription { /// [subscribe]'s `retry`. TLS (and skipping cert verification via a /// `?tlsInsecure=true` query flag) is derived from the endpoint, so the whole connection /// is described in one place. `subscribe` opens the connection lazily and returns a -/// [PushSubscription] handle. The credential is read off the client — set a JWT or session -/// on it (`Client.setJWT` / `Client.setSession`), the same way every other service reads -/// auth. The broker location comes from `Client.setPushEndpoint()` (or the client +/// [PushSubscription] handle. The credential is read off the client: a JWT or session set on +/// it (`Client.setJWT` / `Client.setSession`), else the session the app signed in with through +/// it (e.g. `Account.createEmailPasswordSession`). The broker location comes from `Client.setPushEndpoint()` (or the client /// endpoint) and a stable id from `Client.setPushClientId()`. /// /// ```dart @@ -86,8 +149,8 @@ abstract class Push extends Service { /// The future resolves only once every subscription has been acknowledged (SUBACK), so a /// subscribe-then-publish is reliable, and [onOpen]/[onClose]/[onError] observe the /// in-process connection. On Android a [background] subscription moves the connection to the - /// SDK's native plugin, which also resolves after SUBACK and forwards messages and errors (to - /// [onError]) but not its connection lifecycle, so [onOpen]/[onClose] do not fire for it. On + /// SDK's native plugin, which also resolves after SUBACK and forwards messages, errors (to + /// [onError]) and its connection opening and closing (to [onOpen]/[onClose]). On /// iOS it hands the connection to a background isolate, which subscribes asynchronously and /// retries failures internally, so it resolves once the isolate has the subscription. /// @@ -97,8 +160,10 @@ abstract class Push extends Service { /// always keeps its session, so replay is a purely per-subscription choice. /// /// Set [background] to also keep receiving while the app is backgrounded or closed and post a - /// notification per message with [title] (defaulting to the message topic). Toggle these - /// later with [PushSubscription.update]. + /// notification per message with [title] (defaulting to the message topic). On Android and the + /// web, notifications are posted only while the app is not on screen, since it shows the + /// message itself through the callback; set [notifyInForeground] to post them while it is too. + /// Toggle these later with [PushSubscription.update]. /// /// On Android the SDK's native plugin saves the subscription and keeps delivering after the /// app is killed, the device restarts or the app updates, until it is unsubscribed or [close] @@ -114,6 +179,7 @@ abstract class Push extends Service { bool? background, String? title, bool retry, + bool notifyInForeground, }); /// Android: run background delivery in a foreground service (with a quiet ongoing @@ -123,6 +189,43 @@ abstract class Push extends Service { /// background. It only affects [background] subscriptions. A no-op elsewhere. Future setForeground(bool enabled); + /// Android: what background delivery can rely on. When `bestEffort` is true the scheduled + /// wake-ups are inexact and Doze can defer them, so messages may arrive late while the app is + /// closed; explain why, then ask with [requestExactAlarms] or + /// [requestIgnoreBatteryOptimizations]. Null elsewhere. + Future backgroundStatus(); + + /// Android 12+: open the system screen where the user allows exact alarms, for punctual + /// background wake-ups. Call it from a user action, never on its own. Resolves false when + /// there is nothing to ask: already allowed, older Android, the app does not declare + /// `SCHEDULE_EXACT_ALARM`, or not Android. + Future requestExactAlarms(); + + /// Android: ask the user to exempt the app from battery optimisation. Call it from a user + /// action. Resolves false when there is nothing to ask: already exempt, the app does not + /// declare `REQUEST_IGNORE_BATTERY_OPTIMIZATIONS`, or not Android. + Future requestIgnoreBatteryOptimizations(); + + /// The background notification whose tap launched the app, or null. Reported once; call it + /// when the app starts, e.g. to open the screen the message is about. + /// + /// ```dart + /// final opened = await push.getInitialNotification(); + /// if (opened != null) openSale(opened.data['saleId']); + /// ``` + Future getInitialNotification(); + + /// Call [callback] each time the user taps a background notification while the app is + /// running, including in the background. Returns a function that stops listening. The tap + /// that launched the app comes from [getInitialNotification] instead. + /// + /// ```dart + /// final stop = push.onNotificationOpened((opened) => openSale(opened.data['saleId'])); + /// ``` + void Function() onNotificationOpened( + void Function(PushNotificationOpened opened) callback, + ); + /// Tear down the connection and drop all subscriptions. On Android this also stops /// background delivery, including subscriptions saved by an earlier run: call it on sign-out. void close(); diff --git a/lib/src/mqtt_browser.dart b/lib/src/mqtt_browser.dart index dfba4b78..81041d43 100644 --- a/lib/src/mqtt_browser.dart +++ b/lib/src/mqtt_browser.dart @@ -22,6 +22,7 @@ class _Subscription { bool background; String? title; bool retry; + bool notifyInForeground; _Subscription( this.filter, @@ -29,12 +30,19 @@ class _Subscription { this.background = false, this.title, this.retry = true, + this.notifyInForeground = false, }); } class _SubscriptionHandle implements PushSubscription { final void Function() _unsubscribe; - final void Function({bool? background, String? title, bool? retry}) _update; + final void Function({ + bool? background, + String? title, + bool? retry, + bool? notifyInForeground, + }) + _update; _SubscriptionHandle(this._unsubscribe, this._update); @@ -42,8 +50,17 @@ class _SubscriptionHandle implements PushSubscription { void unsubscribe() => _unsubscribe(); @override - void update({bool? background, String? title, bool? retry}) => - _update(background: background, title: title, retry: retry); + void update({ + bool? background, + String? title, + bool? retry, + bool? notifyInForeground, + }) => _update( + background: background, + title: title, + retry: retry, + notifyInForeground: notifyInForeground, + ); } // Fixed connection tuning — not exposed as an option. @@ -251,6 +268,7 @@ class PushWeb implements Push { bool? background, String? title, bool retry = true, + bool notifyInForeground = false, }) async { final List topicList; final MqttBrowserClient mqtt; @@ -265,6 +283,7 @@ class PushWeb implements Push { background: background, title: title, retry: retry, + notifyInForeground: notifyInForeground, ); } _notify(e); @@ -337,6 +356,7 @@ class PushWeb implements Push { background: wantsBackground, title: title, retry: retry, + notifyInForeground: notifyInForeground, ); final completer = Completer(); @@ -361,6 +381,7 @@ class PushWeb implements Push { background: background, title: title, retry: retry, + notifyInForeground: notifyInForeground, ); } unsubscribe(); @@ -368,7 +389,12 @@ class PushWeb implements Push { rethrow; } - void update({bool? background, String? title, bool? retry}) { + void update({ + bool? background, + String? title, + bool? retry, + bool? notifyInForeground, + }) { final changedFilters = {}; for (final id in ids) { final entry = _subscriptions[id]; @@ -381,6 +407,9 @@ class PushWeb implements Push { if (title != null) { entry.title = title; } + if (notifyInForeground != null) { + entry.notifyInForeground = notifyInForeground; + } if (retry != null && retry != entry.retry) { entry.retry = retry; changedFilters.add(entry.filter); @@ -424,6 +453,30 @@ class PushWeb implements Push { @override Future setForeground(bool enabled) async {} + @override + Future backgroundStatus() async => null; + + @override + Future requestExactAlarms() async => false; + + @override + Future requestIgnoreBatteryOptimizations() async => false; + + // Taps on the notifications this page shows, while it is open. + static final StreamController _opened = + StreamController.broadcast(); + + @override + Future getInitialNotification() async => null; + + @override + void Function() onNotificationOpened( + void Function(PushNotificationOpened opened) callback, + ) { + final subscription = _opened.stream.listen(callback); + return () => unawaited(subscription.cancel()); + } + @override void close() { _mqtt?.disconnect(); @@ -474,10 +527,11 @@ class PushWeb implements Push { return connecting.then((_) => _mqtt!); } - // The credential set on the client (via Client.setJWT / setSession), as (method, credential). + // The credential set on the client (via Client.setJWT / setSession), else the session it + // signed in with, as (method, credential). (String, String) _credential() { final jwt = client.config['jwt'] ?? client.config['jWT']; - final session = client.config['session']; + final session = _clientSession(client); if (jwt != null && jwt.isNotEmpty) { return ('appwrite-jwt', jwt); } @@ -485,7 +539,7 @@ class PushWeb implements Push { return ('appwrite-session', session); } throw AppwriteException( - 'No credential set on the client; call Client.setJWT() or Client.setSession() first.', + 'No credential: sign in with this client, or call Client.setJWT() or Client.setSession() first.', ); } @@ -619,22 +673,30 @@ class PushWeb implements Push { qos: publish.header?.qos.index ?? 0, ); final content = PushNotificationContent.of(message); - final titles = {}; + final shownTitles = {}; for (final subscription in _subscriptions.values) { if (_matches(subscription.filter, topic)) { subscription.callback(message); - // Notification is per-subscription: only subs that opted in show one, each with - // its own title. A title the server sent replaces theirs, so one is shown. - if (subscription.background) { - titles.add(content.titleOr(subscription.title ?? topic)); + // Notification is per-subscription: only subs that opted in show one, each + // title once. A title the server sent replaces theirs, so it is shown once. + final title = content.titleOr(subscription.title ?? topic); + if (subscription.background && + (html.document.visibilityState != 'visible' || + subscription.notifyInForeground) && + html.Notification.supported && + html.Notification.permission == 'granted' && + shownTitles.add(title)) { + html.Notification( + title, + body: content.bodyFor(message), + ).onClick.listen((_) { + _opened.add( + PushNotificationOpened.fromPayload(topic, message.data), + ); + }); } } } - if (html.Notification.permission == 'granted') { - for (final title in titles) { - html.Notification(title, body: content.bodyFor(message)); - } - } } } @@ -856,11 +918,29 @@ List _topicList(Client client, Object? topics) { : [topics.toString()]; } +/// The session set on the client, else the one it signed in with when the browser keeps it in +/// localStorage (`cookieFallback`). +String? _clientSession(Client client) { + final session = client.config['session']; + if (session != null && session.isNotEmpty) { + return session; + } + try { + final fallback = + jsonDecode(html.window.localStorage['cookieFallback'] ?? '{}') + as Map; + final value = fallback['a_session_${client.config['project'] ?? ''}']; + return value is String && value.isNotEmpty ? value : null; + } catch (_) { + return null; + } +} + /// The signed-in user's own topic, `users/`, read off the client's /// credential (JWT first, then session) without a network call. String _userTopic(Client client) { final jwt = client.config['jwt'] ?? client.config['jWT']; - final session = client.config['session']; + final session = _clientSession(client); // The credential the connection authenticates with: the JWT when one is set (never the // session, which could belong to a different user), else the session. final userId = @@ -869,7 +949,7 @@ String _userTopic(Client client) { : _userIdFromSession(session); if (userId == null) { throw AppwriteException( - 'subscribe() without a topic needs a signed-in user: set a JWT or session on the client', + 'subscribe() without a topic needs a signed-in user: sign in with this client, or set a JWT or session on it', ); } return 'users/$userId'; diff --git a/lib/src/mqtt_foreground.dart b/lib/src/mqtt_foreground.dart index 7086bf08..f1869fec 100644 --- a/lib/src/mqtt_foreground.dart +++ b/lib/src/mqtt_foreground.dart @@ -42,6 +42,66 @@ const Duration _retryDelay = Duration(seconds: 2); /// The service event the app listens on to forward background messages to callbacks. String get pushForegroundMessageEvent => _messageEvent; +// Notification taps in the app isolate (iOS): each notification the background isolate posts +// carries its message's topic and payload, read back when it is tapped. +final FlutterLocalNotificationsPlugin _appNotifications = + FlutterLocalNotificationsPlugin(); +final StreamController _localTaps = + StreamController.broadcast(); +Future? _localTapsReady; +bool _launchTapReported = false; + +PushNotificationOpened? _openedFrom(String? payload) { + try { + final tap = jsonDecode(payload ?? '') as Map; + return PushNotificationOpened.fromPayload( + tap['topic'] as String, + tap['payload'] as String, + ); + } catch (_) { + return null; + } +} + +Future _ensureLocalTaps() => + _localTapsReady ??= _appNotifications + .initialize( + const InitializationSettings(iOS: DarwinInitializationSettings()), + onDidReceiveNotificationResponse: (response) { + final opened = _openedFrom(response.payload); + if (opened != null) { + _localTaps.add(opened); + } + }, + ) + .then((_) {}); + +/// iOS: the tap on a background notification that launched the app, once, or null. +Future localNotificationLaunch() async { + if (!Platform.isIOS || _launchTapReported) { + return null; + } + _launchTapReported = true; + await _ensureLocalTaps(); + final details = await _appNotifications.getNotificationAppLaunchDetails(); + return details?.didNotificationLaunchApp == true + ? _openedFrom(details!.notificationResponse?.payload) + : null; +} + +/// iOS: call [callback] for each tap on a background notification. Returns a function that +/// stops listening. +void Function() listenLocalNotificationTaps( + void Function(PushNotificationOpened opened) callback, +) { + if (!Platform.isIOS) { + return () {}; + } + unawaited(_ensureLocalTaps()); + final subscription = _localTaps.stream.listen(callback); + return () => unawaited(subscription.cancel()); +} + /// The service event the isolate relays its errors on, so the app's onError sees them. String get pushForegroundErrorEvent => _errorEvent; @@ -263,6 +323,10 @@ Future pushForegroundEntry(ServiceInstance service) async { ), iOS: DarwinNotificationDetails(), ), + payload: jsonEncode({ + 'topic': message.topic, + 'payload': message.data, + }), ); } // Forward to the app isolate (if alive) so in-app callbacks still fire. diff --git a/lib/src/mqtt_io.dart b/lib/src/mqtt_io.dart index a213fc0a..43cd693a 100644 --- a/lib/src/mqtt_io.dart +++ b/lib/src/mqtt_io.dart @@ -8,6 +8,7 @@ import 'package:mqtt5_client/mqtt5_client.dart'; import 'package:mqtt5_client/mqtt5_server_client.dart'; import 'package:typed_data/typed_data.dart' as typed; import 'client.dart'; +import 'client_io.dart'; import 'exception.dart'; import 'mqtt.dart'; import 'mqtt_foreground.dart'; @@ -22,6 +23,7 @@ class _Subscription { bool background; String? title; bool retry; + bool notifyInForeground; _Subscription( this.filter, @@ -29,12 +31,19 @@ class _Subscription { this.background = false, this.title, this.retry = true, + this.notifyInForeground = false, }); } class _SubscriptionHandle implements PushSubscription { final void Function() _unsubscribe; - final void Function({bool? background, String? title, bool? retry}) _update; + final void Function({ + bool? background, + String? title, + bool? retry, + bool? notifyInForeground, + }) + _update; _SubscriptionHandle(this._unsubscribe, this._update); @@ -42,8 +51,17 @@ class _SubscriptionHandle implements PushSubscription { void unsubscribe() => _unsubscribe(); @override - void update({bool? background, String? title, bool? retry}) => - _update(background: background, title: title, retry: retry); + void update({ + bool? background, + String? title, + bool? retry, + bool? notifyInForeground, + }) => _update( + background: background, + title: title, + retry: retry, + notifyInForeground: notifyInForeground, + ); } // Fixed connection tuning — not exposed as an option. @@ -84,9 +102,13 @@ class PushIO implements Push { final PushNative? _native = PushNative.instance; // Whether the subscriptions currently live on the native plugin's connection. bool _nativeActive = false; + bool _nativeOpen = false; // Bumped when the subscriptions move to the native host, or the credential changes, // superseding an _open in flight. int _connectionEpoch = 0; + + // Bumped by close(), so work that awaited across it does not subscribe or resume afterwards. + int _closeGeneration = 0; Completer? _openCancel; String _connectedKey = ''; bool _resubscribeOnConnect = false; @@ -156,8 +178,55 @@ class PushIO implements Push { _tls = tls, _tlsInsecure = tlsInsecure { // Resume background delivery saved by an earlier run now, instead of at its next - // scheduled wake-up. - unawaited(_native?.resume().catchError(_report)); + // scheduled wake-up, with the credential set on the client: a rotated session of the same + // user replaces the saved one, and another user drops the saved subscriptions. + final native = _native; + if (native != null) { + final generation = _closeGeneration; + unawaited( + _readCookieSession() + .then((_) { + if (generation != _closeGeneration) { + return null; + } + final (authMethod, credential) = _currentCredential(); + return native.resume(authMethod, credential); + }) + .catchError(_report), + ); + } + } + + // The session the app signed in with (an email, OAuth or other sign-in through this client), + // read from the client's cookie store; used when no JWT or session was set on the client. + String? _cookieSession; + + Future _readCookieSession() async { + final client = this.client; + if (client is! ClientIO) { + return; + } + try { + // Wait for an initialization in progress, and initialize only a client that never was. + while (!client.initialized && client.initProgress) { + await Future.delayed(const Duration(milliseconds: 10)); + } + if (!client.initialized) { + await client.init(); + } + final name = 'a_session_${client.config['project'] ?? ''}'; + final cookies = await client.cookieJar.loadForRequest( + Uri.parse(client.endPoint), + ); + final session = + cookies + .where((cookie) => cookie.name == name && cookie.value.isNotEmpty) + .map((cookie) => Uri.decodeComponent(cookie.value)) + .firstOrNull; + _cookieSession = session; + } catch (_) { + // No cookie store yet: the client has not signed in or made a request. + } } // Empty when the app did not set one: the broker derives a stable id server-side (keyed on @@ -250,11 +319,58 @@ class PushIO implements Push { return this; } + @override + Future backgroundStatus() async { + final json = await _native?.backgroundStatus(); + return json == null + ? null + : PushBackgroundStatus.fromMap( + jsonDecode(json) as Map, + ); + } + + @override + Future requestExactAlarms() async => + await _native?.requestExactAlarms() ?? false; + + @override + Future requestIgnoreBatteryOptimizations() async => + await _native?.requestIgnoreBatteryOptimizations() ?? false; + @override Future setForeground(bool enabled) async { await _native?.setForeground(enabled); } + @override + Future getInitialNotification() async { + final native = _native; + if (native == null) { + return localNotificationLaunch(); + } + final tap = await native.getInitialNotification(); + return tap == null + ? null + : PushNotificationOpened.fromPayload( + tap['topic'] as String, + tap['payload'] as String, + ); + } + + @override + void Function() onNotificationOpened( + void Function(PushNotificationOpened opened) callback, + ) { + final native = _native; + if (native == null) { + return listenLocalNotificationTaps(callback); + } + return native.listenOpened( + (topic, payload) => + callback(PushNotificationOpened.fromPayload(topic, payload)), + ); + } + // Errors already handed to onError, so one failure reaching it through two paths (a // refused CONNACK that also fails the subscribe waiting on it) is reported once. final Expando _reported = Expando(); @@ -308,10 +424,18 @@ class PushIO implements Push { bool? background, String? title, bool retry = true, + bool notifyInForeground = false, }) async { + final generation = _closeGeneration; + await _readCookieSession(); + if (generation != _closeGeneration) { + throw AppwriteException( + 'Push was closed before the subscription was established', + ); + } final List topicList; try { - topicList = _topicList(client, topics); + topicList = _topicList(client, topics, _cookieSession); } catch (e) { _notify(e); rethrow; @@ -330,6 +454,7 @@ class PushIO implements Push { background: wantsBackground, title: title, retry: retry, + notifyInForeground: notifyInForeground, ); } @@ -352,6 +477,11 @@ class PushIO implements Push { await _subscribeInProcess(ids); break; } on _Superseded { + if (generation != _closeGeneration) { + throw AppwriteException( + 'Push was closed before the subscription was established', + ); + } // Another Push moved this one's subscriptions to the native host meanwhile: this // subscribe goes there too. if (_nativeActive) { @@ -373,8 +503,18 @@ class PushIO implements Push { return _SubscriptionHandle( () => _unsubscribeIds(ids), - ({bool? background, String? title, bool? retry}) => - _updateIds(ids, background: background, title: title, retry: retry), + ({ + bool? background, + String? title, + bool? retry, + bool? notifyInForeground, + }) => _updateIds( + ids, + background: background, + title: title, + retry: retry, + notifyInForeground: notifyInForeground, + ), ); } @@ -429,18 +569,35 @@ class PushIO implements Push { } } _joinNative(native); - return _enqueueNative(() => _sendToNative(native)); + // Instances that joined an open connection hear onOpen now; the others when it opens. + return _enqueueNative(() => _sendToNative(native)).then((open) { + if (open) { + for (final push in _nativeHosts.toList()) { + push._nativeConnection(true); + } + } + }); } - // Host every native host's subscriptions with this Push's credential. - Future _sendToNative(PushNative native) async { + // This Push's view of the native host's connection, reported once per change. + void _nativeConnection(bool open) { + if (!_nativeActive || _nativeOpen == open) { + return; + } + _nativeOpen = open; + open ? _onOpen?.call() : _onClose?.call(); + } + + // Host every native host's subscriptions with this Push's credential. Completes with whether the + // connection is open (false when there was nothing to host). + Future _sendToNative(PushNative native) async { // Host what is subscribed when this runs, so a close() or an unsubscribe queued meanwhile // is not undone by an older request. final subscriptions = [ for (final push in _nativeHosts) ...push._nativeEntries, ]; if (subscriptions.isEmpty) { - return; + return false; } if (!_notificationPermissionRequested && _nativeHosts.any((push) => push._backgroundWanted)) { @@ -467,7 +624,7 @@ class PushIO implements Push { await native.setErrorCallback( _nativeHosts.any((push) => push._onError != null), ); - await native.host(jsonEncode(config), jsonEncode(subscriptions)); + return native.host(jsonEncode(config), jsonEncode(subscriptions)); } // Move this Push's subscriptions onto the native host, closing its own connection. @@ -496,8 +653,12 @@ class PushIO implements Push { unawaited(native.ack(message.ackToken).catchError(_report)); }, onError: (message) => _report(AppwriteException(message)), + onConnection: _nativeConnection, ); _nativeHosts.add(this); + if (!_nativeActive) { + _nativeOpen = false; + } _nativeActive = true; } @@ -553,6 +714,7 @@ class PushIO implements Push { 'background': entry.value.background, 'title': entry.value.title, 'retry': entry.value.retry, + 'notifyInForeground': entry.value.notifyInForeground, }, ]; @@ -834,6 +996,7 @@ class PushIO implements Push { bool? background, String? title, bool? retry, + bool? notifyInForeground, }) { for (final id in ids) { final sub = _subscriptions[id]; @@ -849,6 +1012,9 @@ class PushIO implements Push { if (retry != null) { sub.retry = retry; } + if (notifyInForeground != null) { + sub.notifyInForeground = notifyInForeground; + } } // reconcile re-subscribes (in-process) / re-syncs the isolate at the new effective QoS. _reconcile(); @@ -860,6 +1026,7 @@ class PushIO implements Push { /// subscriptions and the saved ones. @override void close() { + _closeGeneration++; _teardown(); final native = _native; if (native == null) { @@ -880,6 +1047,10 @@ class PushIO implements Push { // Close the in-process connection (and the isolate host) and drop the live subscriptions. void _teardown() { _livePushes.remove(this); + // A connection still being set up sees the new epoch and gives up instead of connecting. + _connectionEpoch++; + _openCancel?.complete(); + _openCancel = null; _mqtt?.disconnect(); _mqtt = null; _connecting = null; @@ -920,7 +1091,17 @@ class PushIO implements Push { return connecting.then((_) => _mqtt!); } - // The credential set on the client (via Client.setJWT / setSession), as (method, credential). + // The credential set on the client, or (null, null) when there is none. + (String?, String?) _currentCredential() { + try { + return _credential(); + } catch (_) { + return (null, null); + } + } + + // The credential set on the client (via Client.setJWT / setSession), else the session it + // signed in with, as (method, credential). (String, String) _credential() { final jwt = client.config['jwt'] ?? client.config['jWT']; final session = client.config['session']; @@ -930,8 +1111,12 @@ class PushIO implements Push { if (session != null && session.isNotEmpty) { return ('appwrite-session', session); } + final cookieSession = _cookieSession; + if (cookieSession != null && cookieSession.isNotEmpty) { + return ('appwrite-session', cookieSession); + } throw AppwriteException( - 'No credential set on the client; call Client.setJWT() or Client.setSession() first.', + 'No credential: sign in with this client, or call Client.setJWT() or Client.setSession() first.', ); } @@ -1273,9 +1458,9 @@ bool _matches(String filter, String topic) { /// Resolve [Push.subscribe]'s `topics`: a String, Topic or ResolvedTopic, or a List of /// them, or null for the signed-in user's own `users/` topic. -List _topicList(Client client, Object? topics) { +List _topicList(Client client, Object? topics, String? cookieSession) { if (topics == null) { - return [_userTopic(client)]; + return [_userTopic(client, cookieSession)]; } // A topic is a String or a Topic / ResolvedTopic builder (via toString()). return topics is List @@ -1284,10 +1469,13 @@ List _topicList(Client client, Object? topics) { } /// The signed-in user's own topic, `users/`, read off the client's -/// credential (JWT first, then session) without a network call. -String _userTopic(Client client) { +/// credential (JWT first, then session, then the session it signed in with) without a network +/// call. +String _userTopic(Client client, String? cookieSession) { final jwt = client.config['jwt'] ?? client.config['jWT']; - final session = client.config['session']; + final configured = client.config['session']; + final session = + configured != null && configured.isNotEmpty ? configured : cookieSession; // The credential the connection authenticates with: the JWT when one is set (never the // session, which could belong to a different user), else the session. final userId = @@ -1296,7 +1484,7 @@ String _userTopic(Client client) { : _userIdFromSession(session); if (userId == null) { throw AppwriteException( - 'subscribe() without a topic needs a signed-in user: set a JWT or session on the client', + 'subscribe() without a topic needs a signed-in user: sign in with this client, or set a JWT or session on it', ); } return 'users/$userId'; diff --git a/lib/src/mqtt_native.dart b/lib/src/mqtt_native.dart index 2e060782..896ecdd3 100644 --- a/lib/src/mqtt_native.dart +++ b/lib/src/mqtt_native.dart @@ -58,12 +58,13 @@ class PushNative { debugInstance ?? (Platform.isAndroid ? _plugin : null); /// Host [subscriptionsJson] on the background connection described by [configJson]. Completes - /// once the connection is up and every filter is subscribed, or throws why not. - Future host(String configJson, String subscriptionsJson) => - _methods.invokeMethod('host', { + /// with whether the connection is open once every filter is subscribed, or throws why not. + Future host(String configJson, String subscriptionsJson) async => + await _methods.invokeMethod('host', { 'config': configJson, 'subscriptions': subscriptionsJson, - }); + }) ?? + false; /// Ask for the notification permission background messages are posted with (Android 13+), /// without waiting for the answer. False when no Activity was attached to ask from. @@ -89,8 +90,27 @@ class PushNative { Future hasSaved() async => await _methods.invokeMethod('hasSaved') ?? false; - /// Resume saved background delivery now instead of at the next scheduled run. - Future resume() => _methods.invokeMethod('resume'); + /// Resume saved background delivery now instead of at the next scheduled run, with the + /// credential set on the client (null when none is): a rotated session of the same user + /// replaces the saved one, and another user drops the saved subscriptions. + Future resume(String? authMethod, String? credential) => + _methods.invokeMethod('resume', { + 'authMethod': authMethod, + 'credential': credential, + }); + + /// What background delivery can rely on, as JSON. + Future backgroundStatus() => + _methods.invokeMethod('backgroundStatus'); + + /// Open the system screen that allows exact alarms; false when there is nothing to ask. + Future requestExactAlarms() async => + await _methods.invokeMethod('requestExactAlarms') ?? false; + + /// Ask to exempt the app from battery optimisation; false when there is nothing to ask. + Future requestIgnoreBatteryOptimizations() async => + await _methods.invokeMethod('requestIgnoreBatteryOptimizations') ?? + false; /// Record whether onError is registered; returns a refusal kept while it was not, once. Future setErrorCallback(bool registered) => _methods @@ -104,10 +124,48 @@ class PushNative { }) ?? ''; - /// Listen to the plugin's messages and errors. + /// The tapped notification that launched the app as `{topic, payload}`, once, or null. + Future?> getInitialNotification() => + _methods.invokeMethod>('getInitialNotification'); + + /// Call [onOpened] with the topic and payload of each notification tap while listening. + /// Until something listens, a tap waits for [getInitialNotification]. Returns a function + /// that stops listening. + void Function() listenOpened( + void Function(String topic, String payload) onOpened, + ) { + final stream = + _stream ??= + _events + .receiveBroadcastStream() + .map((event) => event as Map) + .asBroadcastStream(); + final subscription = stream.listen((event) { + if (event['type'] == 'opened') { + onOpened(event['topic'] as String, event['payload'] as String); + } + }); + unawaited( + _methods.invokeMethod('listenOpened', {'listening': true}), + ); + var listening = true; + return () { + if (!listening) { + return; + } + listening = false; + unawaited(subscription.cancel()); + unawaited( + _methods.invokeMethod('listenOpened', {'listening': false}), + ); + }; + } + + /// Listen to the plugin's messages, errors and connection changes. StreamSubscription> listen({ required void Function(NativePushMessage message) onMessage, required void Function(String message) onError, + required void Function(bool connected) onConnection, }) { final stream = _stream ??= @@ -128,6 +186,8 @@ class PushNative { ); } else if (event['type'] == 'error') { onError(event['message'] as String? ?? 'Background push failed'); + } else if (event['type'] == 'connection') { + onConnection(event['connected'] == true); } }); } diff --git a/pubspec.yaml b/pubspec.yaml index 78cf462f..c3e3eee8 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -1,5 +1,5 @@ name: appwrite -version: 27.1.0-rc.3 +version: 27.1.0-rc.7 description: Appwrite is an open-source self-hosted backend server that abstracts and simplifies complex and repetitive development tasks behind a very simple REST API homepage: https://appwrite.io repository: https://github.com/appwrite/sdk-for-flutter