Wait for relay OK ack before marking SENT

- Quartz: expose pendingPublishRelaysFor(eventId) on INostrClient - Worker: after publish(), poll pendingPublishRelaysFor every 500ms up to OK_TIMEOUT_SEC=30s - notify user when a scheduled post fires or fails
2026-05-07 16:20:10 +02:00
parent 28defaf96e
commit f6991bee15
10 changed files with 219 additions and 2 deletions
@@ -0,0 +1,138 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.service.scheduledposts
+
+import android.app.NotificationChannel
+import android.app.NotificationManager
+import android.app.PendingIntent
+import android.content.Context
+import android.content.Intent
+import androidx.core.app.NotificationCompat
+import androidx.core.app.NotificationManagerCompat
+import com.vitorpamplona.amethyst.R
+import com.vitorpamplona.amethyst.ui.MainActivity
+import com.vitorpamplona.amethyst.ui.stringRes
+import com.vitorpamplona.quartz.nip01Core.core.Event
+
+/**
+ * Posts user-visible system notifications when a scheduled post completes
+ * (sent or failed). Without this, a worker firing in background offers zero
+ * diagnostic to the user — see the silent-publish bug the ack-aware worker
+ * now guards against; the notification closes the loop.
+ */
+object ScheduledPostNotifier {
+    private var channel: NotificationChannel? = null
+    private const val SCHEDULED_POST_NOT_ID_BASE = 0x70000
+
+    fun notifySent(
+        context: Context,
+        post: ScheduledPost,
+    ) {
+        ensureChannel(context)
+        post(
+            context = context,
+            notId = idFor(post.id),
+            title = stringRes(context, R.string.scheduled_posts_notification_sent_title),
+            body = previewOf(post),
+        )
+    }
+
+    fun notifyFailed(
+        context: Context,
+        post: ScheduledPost,
+        error: String?,
+    ) {
+        ensureChannel(context)
+        val snippet = previewOf(post)
+        val body =
+            if (error.isNullOrBlank()) {
+                snippet
+            } else {
+                "$snippet\n${stringRes(context, R.string.scheduled_posts_error_prefix, error)}"
+            }
+        post(
+            context = context,
+            notId = idFor(post.id),
+            title = stringRes(context, R.string.scheduled_posts_notification_failed_title),
+            body = body,
+        )
+    }
+
+    private fun post(
+        context: Context,
+        notId: Int,
+        title: String,
+        body: String,
+    ) {
+        val channelId = stringRes(context, R.string.app_notification_scheduled_posts_channel_id)
+        val tapIntent =
+            Intent(context, MainActivity::class.java).apply {
+                addFlags(Intent.FLAG_ACTIVITY_NEW_TASK or Intent.FLAG_ACTIVITY_CLEAR_TOP)
+            }
+        val tapPendingIntent =
+            PendingIntent.getActivity(
+                context,
+                notId,
+                tapIntent,
+                PendingIntent.FLAG_IMMUTABLE or PendingIntent.FLAG_UPDATE_CURRENT,
+            )
+        val builder =
+            NotificationCompat
+                .Builder(context, channelId)
+                .setSmallIcon(R.drawable.amethyst)
+                .setContentTitle(title)
+                .setContentText(body)
+                .setStyle(NotificationCompat.BigTextStyle().bigText(body))
+                .setContentIntent(tapPendingIntent)
+                .setPriority(NotificationCompat.PRIORITY_DEFAULT)
+                .setCategory(NotificationCompat.CATEGORY_STATUS)
+                .setAutoCancel(true)
+                .setWhen(System.currentTimeMillis())
+        // Silently no-ops on Android 13+ if POST_NOTIFICATIONS isn't granted.
+        NotificationManagerCompat.from(context).notify(notId, builder.build())
+    }
+
+    private fun ensureChannel(context: Context) {
+        if (channel != null) return
+        channel =
+            NotificationChannel(
+                stringRes(context, R.string.app_notification_scheduled_posts_channel_id),
+                stringRes(context, R.string.app_notification_scheduled_posts_channel_name),
+                NotificationManager.IMPORTANCE_DEFAULT,
+            ).apply {
+                description = stringRes(context, R.string.app_notification_scheduled_posts_channel_description)
+            }
+        val nm = context.getSystemService(Context.NOTIFICATION_SERVICE) as NotificationManager
+        nm.createNotificationChannel(channel!!)
+    }
+
+    private fun previewOf(post: ScheduledPost): String =
+        runCatching {
+            Event
+                .fromJson(post.signedEventJson)
+                .content
+                .take(120)
+                .trim()
+        }.getOrDefault("")
+
+    // Distinct id per post so multiple completions don't collapse onto one row.
+    private fun idFor(postId: String): Int = SCHEDULED_POST_NOT_ID_BASE xor postId.hashCode()
+}
@@ -32,9 +32,11 @@ import androidx.work.WorkManager
 import androidx.work.WorkerParameters
 import com.vitorpamplona.amethyst.Amethyst
 import com.vitorpamplona.quartz.nip01Core.core.Event
+import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
 import com.vitorpamplona.quartz.utils.Log
 import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.delay
 import java.util.concurrent.TimeUnit

 /**
@@ -60,6 +62,8 @@ class ScheduledPostWorker(
        private const val TAG = "ScheduledPostWorker"
        private const val WORK_NAME = "scheduled_post_worker"
        private const val WORK_NAME_CATCH_UP = "scheduled_post_worker_catch_up"
+        private const val OK_TIMEOUT_SEC = 30L
+        private const val OK_POLL_MS = 500L

        fun schedule(context: Context) {
            val constraints =
@@ -151,13 +155,23 @@ class ScheduledPostWorker(
                    account.client.publish(event, relays)
                    account.consumePostEvent(event, relays, extras)

+                    val acks = waitForOk(account.client, event.id, relays.size)
+                    if (acks > 0) {
                        store.markSent(post.id)
-                    Log.d(TAG) { "client.publish(${post.id}) done; marked SENT" }
+                        ScheduledPostNotifier.notifySent(applicationContext, post)
+                        Log.d(TAG) { "client.publish(${post.id}) acked by $acks/${relays.size}; marked SENT" }
+                    } else {
+                        val msg = "no relay acknowledged within ${OK_TIMEOUT_SEC}s"
+                        store.markFailed(post.id, msg)
+                        ScheduledPostNotifier.notifyFailed(applicationContext, post, msg)
+                        Log.w(TAG, "client.publish(${post.id}) failed: $msg")
+                    }
                } catch (e: CancellationException) {
                    throw e
                } catch (e: Exception) {
                    Log.e(TAG, "Failed to publish scheduled post ${post.id}", e)
                    store.markFailed(post.id, e.message)
+                    ScheduledPostNotifier.notifyFailed(applicationContext, post, e.message)
                }
            }

@@ -170,4 +184,29 @@ class ScheduledPostWorker(
            Result.retry()
        }
    }
+
+    /**
+     * Polls [INostrClient.pendingPublishRelaysFor] until at least one relay sends
+     * an OK ack or [OK_TIMEOUT_SEC] elapses. Returns the number of relays that
+     * acked. The publish call itself is fire-and-forget; without this wait the
+     * worker can be torn down before the websocket finishes delivery.
+     */
+    private suspend fun waitForOk(
+        client: INostrClient,
+        eventId: String,
+        totalRelays: Int,
+    ): Int {
+        val deadline = System.currentTimeMillis() + OK_TIMEOUT_SEC * 1000
+        while (System.currentTimeMillis() < deadline) {
+            val pending = client.pendingPublishRelaysFor(eventId)
+            // null means the outbox dropped the entry — every relay either OK'd
+            // or hit the discard cap (replaced/pow/deleted/invalid). Treat as full ack.
+            if (pending == null) return totalRelays
+            val acked = totalRelays - pending.size
+            if (acked > 0) return acked
+            delay(OK_POLL_MS)
+        }
+        val pending = client.pendingPublishRelaysFor(eventId)
+        return if (pending == null) totalRelays else totalRelays - pending.size
+    }
 }
@@ -2484,4 +2484,8 @@
    <string name="scheduled_posts_status_sent">Odesláno</string>
    <string name="scheduled_posts_status_cancelled">Zrušeno</string>
    <string name="scheduled_posts_logout_warning">Máte %1$d naplánovaných příspěvků, které ještě nebyly publikovány. Odhlášením budou trvale smazány.</string>
+    <string name="scheduled_posts_notification_sent_title">Naplánovaný příspěvek publikován</string>
+    <string name="scheduled_posts_notification_failed_title">Naplánovaný příspěvek selhal</string>
+    <string name="app_notification_scheduled_posts_channel_name">Naplánované příspěvky</string>
+    <string name="app_notification_scheduled_posts_channel_description">Oznámení, když je naplánovaný příspěvek publikován nebo selže při publikaci.</string>
 </resources>
@@ -2475,4 +2475,8 @@ anz der Bedingungen ist erforderlich</string>
    <string name="scheduled_posts_status_sent">Gesendet</string>
    <string name="scheduled_posts_status_cancelled">Abgebrochen</string>
    <string name="scheduled_posts_logout_warning">Du hast %1$d geplante(n) Beitrag/Beiträge, der/die noch nicht veröffentlicht wurden. Beim Abmelden werden sie dauerhaft gelöscht.</string>
+    <string name="scheduled_posts_notification_sent_title">Geplanter Beitrag veröffentlicht</string>
+    <string name="scheduled_posts_notification_failed_title">Geplanter Beitrag fehlgeschlagen</string>
+    <string name="app_notification_scheduled_posts_channel_name">Geplante Beiträge</string>
+    <string name="app_notification_scheduled_posts_channel_description">Benachrichtigungen, wenn ein geplanter Beitrag veröffentlicht wird oder die Veröffentlichung fehlschlägt.</string>
 </resources>
@@ -2470,4 +2470,8 @@
    <string name="scheduled_posts_status_sent">Enviado</string>
    <string name="scheduled_posts_status_cancelled">Cancelado</string>
    <string name="scheduled_posts_logout_warning">Você tem %1$d post(s) agendado(s) que ainda não foram publicados. Sair excluirá esses posts permanentemente.</string>
+    <string name="scheduled_posts_notification_sent_title">Post agendado publicado</string>
+    <string name="scheduled_posts_notification_failed_title">Post agendado falhou</string>
+    <string name="app_notification_scheduled_posts_channel_name">Posts agendados</string>
+    <string name="app_notification_scheduled_posts_channel_description">Notificações quando um post agendado é publicado ou falha ao publicar.</string>
 </resources>
@@ -2469,4 +2469,8 @@
    <string name="scheduled_posts_status_sent">Skickat</string>
    <string name="scheduled_posts_status_cancelled">Avbrutet</string>
    <string name="scheduled_posts_logout_warning">Du har %1$d schemalagda inlägg som inte har publicerats än. Att logga ut raderar dem permanent.</string>
+    <string name="scheduled_posts_notification_sent_title">Schemalagt inlägg publicerat</string>
+    <string name="scheduled_posts_notification_failed_title">Schemalagt inlägg misslyckades</string>
+    <string name="app_notification_scheduled_posts_channel_name">Schemalagda inlägg</string>
+    <string name="app_notification_scheduled_posts_channel_description">Aviseringar när ett schemalagt inlägg publiceras eller misslyckas att publicera.</string>
 </resources>
@@ -455,6 +455,11 @@
    <string name="scheduled_posts_day_tomorrow">Tomorrow</string>
    <string name="scheduled_posts_logout_toast_zero">Logged out</string>
    <string name="scheduled_posts_logout_toast">Logged out · %1$d scheduled post(s) deleted</string>
+    <string name="scheduled_posts_notification_sent_title">Scheduled post published</string>
+    <string name="scheduled_posts_notification_failed_title">Scheduled post failed</string>
+    <string name="app_notification_scheduled_posts_channel_id" translatable="false">ScheduledPostsID</string>
+    <string name="app_notification_scheduled_posts_channel_name">Scheduled posts</string>
+    <string name="app_notification_scheduled_posts_channel_description">Notifications when a scheduled post is published or fails to publish.</string>
    <string name="scheduled_posts_send_now_title">Send now?</string>
    <string name="scheduled_posts_send_now_message">This post will publish to relays immediately. The original schedule will be discarded.</string>
    <string name="scheduled_posts_send_now_confirm">Send</string>
@@ -72,6 +72,13 @@ interface INostrClient : AutoCloseable {
        relayList: Set<NormalizedRelayUrl>,
    )

+    /**
+     * Returns the relays that have not yet acknowledged [eventId] with an OK,
+     * or null if the event is not tracked (never published, or already fully done).
+     * Use to poll for delivery confirmation after [publish].
+     */
+    fun pendingPublishRelaysFor(eventId: HexKey): Set<NormalizedRelayUrl>?
+
    fun addConnectionListener(listener: RelayConnectionListener)

    fun removeConnectionListener(listener: RelayConnectionListener)
@@ -123,6 +130,8 @@ class EmptyNostrClient : INostrClient {
        relayList: Set<NormalizedRelayUrl>,
    ) { }

+    override fun pendingPublishRelaysFor(eventId: HexKey): Set<NormalizedRelayUrl>? = null
+
    override fun addConnectionListener(listener: RelayConnectionListener) {}

    override fun removeConnectionListener(listener: RelayConnectionListener) {}
@@ -358,6 +358,8 @@ class NostrClient(

    override fun activeOutboxCache(url: NormalizedRelayUrl): Set<HexKey> = eventOutbox.activeOutboxCacheFor(url)

+    override fun pendingPublishRelaysFor(eventId: HexKey): Set<NormalizedRelayUrl>? = eventOutbox.pendingRelaysFor(eventId)
+
    override fun getReqFiltersOrNull(subId: String): Map<NormalizedRelayUrl, List<Filter>>? = activeRequests.getSubscriptionFiltersOrNull(subId)

    override fun getCountFiltersOrNull(subId: String): Map<NormalizedRelayUrl, List<Filter>>? = activeCounts.getSubscriptionFiltersOrNull(subId)
@@ -90,6 +90,14 @@ class PoolEventOutbox {
        return myEvents
    }

+    /**
+     * Returns the relays that have NOT yet acknowledged [eventId] with an OK, or
+     * null if the event is not currently tracked (never sent or already fully done).
+     * Callers can poll this after publish to detect when relays ack: the set shrinks
+     * as OKs arrive, then the entry is removed from the outbox (returns null).
+     */
+    fun pendingRelaysFor(eventId: HexKey): Set<NormalizedRelayUrl>? = eventOutbox[eventId]?.relaysLeft()
+
    fun markAsSending(
        event: Event,
        relays: Set<NormalizedRelayUrl>,