mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-08-11 00:37:41 +00:00
Merge branch 'main' into fix/desktop-log-noise
This commit is contained in:
Róbert Nagy
+38
@@ -0,0 +1,38 @@
|
||||
/*
|
||||
* 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.commons.preview
|
||||
|
||||
/**
|
||||
* Decodes [bytes] into a String using the charset named [charsetName].
|
||||
*
|
||||
* [charsetName] is an IANA charset name (e.g. "UTF-8", "ISO-8859-1",
|
||||
* "windows-1252"). When it is null or cannot be resolved on the current
|
||||
* platform, the implementation falls back to UTF-8.
|
||||
*
|
||||
* The decode is the only platform-specific step of link-preview HTML parsing:
|
||||
* the JVM actual delegates to `java.nio.charset`, which supports every charset
|
||||
* the JRE ships; the iOS actual maps the common web charsets to
|
||||
* `NSStringEncoding` and falls back to UTF-8 for anything exotic.
|
||||
*/
|
||||
expect fun decodeBytes(
|
||||
bytes: ByteArray,
|
||||
charsetName: String?,
|
||||
): String
|
||||
+11
-11
@@ -20,37 +20,37 @@
|
||||
*/
|
||||
package com.vitorpamplona.amethyst.commons.preview
|
||||
|
||||
import java.nio.charset.Charset
|
||||
|
||||
object HtmlCharsetParser {
|
||||
val ATTRIBUTE_VALUE_CHARSET = "charset"
|
||||
val ATTRIBUTE_VALUE_HTTP_EQUIV = "http-equiv"
|
||||
val CONTENT = "content"
|
||||
|
||||
private const val DEFAULT_CHARSET = "UTF-8"
|
||||
|
||||
private val RE_CONTENT_TYPE_CHARSET = Regex("""charset=([^;]+)""")
|
||||
|
||||
fun detectCharset(bodyBytes: ByteArray): Charset {
|
||||
/**
|
||||
* Sniffs the charset declared in the document's `<meta>` tags, returning its
|
||||
* IANA name. Returns [DEFAULT_CHARSET] when no usable declaration is found.
|
||||
*/
|
||||
fun detectCharset(bodyBytes: ByteArray): String {
|
||||
// try to detect charset from meta tags parsed from first 1024 bytes of body
|
||||
val firstPart = String(bodyBytes, 0, 1024, Charset.forName("utf-8"))
|
||||
val firstPart = bodyBytes.decodeToString(0, minOf(1024, bodyBytes.size))
|
||||
val metaTags = MetaTagsParser.parse(firstPart)
|
||||
metaTags.forEach { meta ->
|
||||
val charsetAttr = meta.attr(ATTRIBUTE_VALUE_CHARSET)
|
||||
if (charsetAttr.isNotEmpty()) {
|
||||
runCatching { Charset.forName(charsetAttr) }.getOrNull()?.let {
|
||||
return it
|
||||
}
|
||||
return charsetAttr
|
||||
}
|
||||
if (meta.attr(ATTRIBUTE_VALUE_HTTP_EQUIV).lowercase() == "content-type") {
|
||||
RE_CONTENT_TYPE_CHARSET
|
||||
.find(meta.attr(CONTENT))
|
||||
?.let {
|
||||
runCatching { Charset.forName(it.groupValues[1]) }.getOrNull()
|
||||
}?.let {
|
||||
return it
|
||||
return it.groupValues[1]
|
||||
}
|
||||
}
|
||||
}
|
||||
// defaults to UTF-8
|
||||
return Charset.forName("utf-8")
|
||||
return DEFAULT_CHARSET
|
||||
}
|
||||
}
|
||||
+69
@@ -0,0 +1,69 @@
|
||||
/*
|
||||
* 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.commons.preview
|
||||
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.withContext
|
||||
|
||||
class HtmlParser {
|
||||
companion object {
|
||||
// Byte-order marks mapped to their IANA charset names, longest first so
|
||||
// a 4-byte BOM is matched before a 2-byte one. (Patterns taken from okhttp.)
|
||||
private val UNICODE_BOMS =
|
||||
listOf(
|
||||
byteArrayOf(0x00, 0x00, 0xFF.toByte(), 0xFF.toByte()) to "UTF-32BE",
|
||||
byteArrayOf(0xFF.toByte(), 0xFF.toByte(), 0x00, 0x00) to "UTF-32LE",
|
||||
byteArrayOf(0xEF.toByte(), 0xBB.toByte(), 0xBF.toByte()) to "UTF-8",
|
||||
byteArrayOf(0xFE.toByte(), 0xFF.toByte()) to "UTF-16BE",
|
||||
byteArrayOf(0xFF.toByte(), 0xFE.toByte()) to "UTF-16LE",
|
||||
)
|
||||
}
|
||||
|
||||
suspend fun parseHtml(
|
||||
bodyBytes: ByteArray,
|
||||
charsetName: String?,
|
||||
): Sequence<MetaTag> =
|
||||
withContext(Dispatchers.Default) {
|
||||
// Precedence: explicit charset (e.g. from Content-Type) > BOM >
|
||||
// charset sniffed from <meta> tags (defaults to UTF-8).
|
||||
val name =
|
||||
charsetName
|
||||
?: bodyBytes.bomCharsetName()
|
||||
?: HtmlCharsetParser.detectCharset(bodyBytes)
|
||||
val content = decodeBytes(bodyBytes, name)
|
||||
MetaTagsParser.parse(content)
|
||||
}
|
||||
|
||||
private fun ByteArray.bomCharsetName(): String? {
|
||||
for ((bom, name) in UNICODE_BOMS) {
|
||||
if (startsWith(bom)) return name
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
private fun ByteArray.startsWith(prefix: ByteArray): Boolean {
|
||||
if (size < prefix.size) return false
|
||||
for (i in prefix.indices) {
|
||||
if (this[i] != prefix[i]) return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
}
|
||||
+133
@@ -0,0 +1,133 @@
|
||||
/*
|
||||
* 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.commons.service.broadcast
|
||||
|
||||
import androidx.compose.runtime.Immutable
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.utils.TimeUtils
|
||||
|
||||
/**
|
||||
* Result of a relay's response to an event publish.
|
||||
*/
|
||||
@Immutable
|
||||
sealed class RelayResult {
|
||||
/** Relay accepted the event (OK message with success=true) */
|
||||
data object Success : RelayResult()
|
||||
|
||||
/** Relay rejected the event (OK message with success=false) */
|
||||
data class Error(
|
||||
val message: String,
|
||||
) : RelayResult()
|
||||
|
||||
/** Relay did not respond within timeout */
|
||||
data object Timeout : RelayResult()
|
||||
|
||||
/** Waiting for relay response */
|
||||
data object Pending : RelayResult()
|
||||
|
||||
/** Retry in progress for this relay */
|
||||
data object Retrying : RelayResult()
|
||||
}
|
||||
|
||||
/**
|
||||
* Overall status of a broadcast operation.
|
||||
*/
|
||||
enum class BroadcastStatus {
|
||||
/** Currently waiting for relay responses */
|
||||
IN_PROGRESS,
|
||||
|
||||
/** All relays accepted the event */
|
||||
SUCCESS,
|
||||
|
||||
/** Some relays accepted, some failed */
|
||||
PARTIAL,
|
||||
|
||||
/** No relays accepted the event */
|
||||
FAILED,
|
||||
}
|
||||
|
||||
/**
|
||||
* Tracks a single event broadcast to multiple relays.
|
||||
*/
|
||||
@Immutable
|
||||
data class BroadcastEvent(
|
||||
val id: String,
|
||||
val event: Event,
|
||||
val targetRelays: List<NormalizedRelayUrl>,
|
||||
val startedAt: Long = TimeUtils.now(),
|
||||
val results: Map<NormalizedRelayUrl, RelayResult> = emptyMap(),
|
||||
val status: BroadcastStatus = BroadcastStatus.IN_PROGRESS,
|
||||
) {
|
||||
/** Number of relays that accepted the event */
|
||||
val successCount: Int
|
||||
get() = results.count { it.value is RelayResult.Success }
|
||||
|
||||
/** Number of relays that rejected or timed out */
|
||||
val failureCount: Int
|
||||
get() = results.count { it.value is RelayResult.Error || it.value is RelayResult.Timeout }
|
||||
|
||||
/** Number of relays still pending response */
|
||||
val pendingCount: Int
|
||||
get() = targetRelays.size - results.size
|
||||
|
||||
/** Total number of target relays */
|
||||
val totalRelays: Int
|
||||
get() = targetRelays.size
|
||||
|
||||
/** Progress as a fraction (0.0 to 1.0) */
|
||||
val progress: Float
|
||||
get() = if (totalRelays == 0) 0f else results.size.toFloat() / totalRelays
|
||||
|
||||
/** Whether all relays have responded */
|
||||
val isComplete: Boolean
|
||||
get() = results.size >= targetRelays.size
|
||||
|
||||
/** List of relays that failed and are not currently retrying */
|
||||
val failedRelays: List<NormalizedRelayUrl>
|
||||
get() =
|
||||
results
|
||||
.filter {
|
||||
(it.value is RelayResult.Error || it.value is RelayResult.Timeout) &&
|
||||
it.value !is RelayResult.Retrying
|
||||
}.keys
|
||||
.toList()
|
||||
|
||||
/** List of relays currently being retried */
|
||||
val retryingRelays: List<NormalizedRelayUrl>
|
||||
get() = results.filter { it.value is RelayResult.Retrying }.keys.toList()
|
||||
|
||||
/** Creates a copy with an updated relay result */
|
||||
fun withResult(
|
||||
relay: NormalizedRelayUrl,
|
||||
result: RelayResult,
|
||||
): BroadcastEvent {
|
||||
val newResults = results + (relay to result)
|
||||
val newStatus =
|
||||
when {
|
||||
newResults.size < targetRelays.size -> BroadcastStatus.IN_PROGRESS
|
||||
newResults.all { it.value is RelayResult.Success } -> BroadcastStatus.SUCCESS
|
||||
newResults.none { it.value is RelayResult.Success } -> BroadcastStatus.FAILED
|
||||
else -> BroadcastStatus.PARTIAL
|
||||
}
|
||||
return copy(results = newResults, status = newStatus)
|
||||
}
|
||||
}
|
||||
+402
@@ -0,0 +1,402 @@
|
||||
/*
|
||||
* 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.commons.service.broadcast
|
||||
|
||||
import com.vitorpamplona.quartz.nip01Core.core.Event
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.listeners.RelayConnectionListener
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.client.single.IRelayClient
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.Message
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.commands.toClient.OkMessage
|
||||
import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
|
||||
import com.vitorpamplona.quartz.utils.Log
|
||||
import com.vitorpamplona.quartz.utils.RandomInstance
|
||||
import kotlinx.collections.immutable.ImmutableList
|
||||
import kotlinx.collections.immutable.persistentListOf
|
||||
import kotlinx.collections.immutable.toImmutableList
|
||||
import kotlinx.coroutines.DelicateCoroutinesApi
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
import kotlinx.coroutines.flow.asStateFlow
|
||||
import kotlinx.coroutines.flow.update
|
||||
import kotlinx.coroutines.withTimeoutOrNull
|
||||
|
||||
/**
|
||||
* Tracks event broadcasts to relays with live progress updates.
|
||||
*
|
||||
* Provides:
|
||||
* - Real-time progress as relays respond
|
||||
* - Detailed per-relay success/error information
|
||||
* - Retry functionality for failed relays
|
||||
*/
|
||||
class BroadcastTracker {
|
||||
companion object {
|
||||
private const val TAG = "BroadcastTracker"
|
||||
const val TIMEOUT_SECONDS = 10L
|
||||
}
|
||||
|
||||
private val _activeBroadcasts = MutableStateFlow<ImmutableList<BroadcastEvent>>(persistentListOf())
|
||||
val activeBroadcasts: StateFlow<ImmutableList<BroadcastEvent>> = _activeBroadcasts.asStateFlow()
|
||||
|
||||
/**
|
||||
* Tracks an event broadcast to relays with live progress updates.
|
||||
*
|
||||
* @param event The Nostr event to broadcast
|
||||
* @param relays Target relays to send to
|
||||
* @param client The Nostr client for sending
|
||||
*/
|
||||
@OptIn(DelicateCoroutinesApi::class)
|
||||
suspend fun trackBroadcast(
|
||||
event: Event,
|
||||
relays: Set<NormalizedRelayUrl>,
|
||||
client: INostrClient,
|
||||
) {
|
||||
val trackingId = RandomInstance.randomChars(16)
|
||||
|
||||
val broadcast =
|
||||
BroadcastEvent(
|
||||
id = trackingId,
|
||||
event = event,
|
||||
targetRelays = relays.toList(),
|
||||
)
|
||||
|
||||
// Add to active broadcasts and cache event for retries
|
||||
_activeBroadcasts.update { (it + broadcast).toImmutableList() }
|
||||
|
||||
Log.d(TAG) { "Starting broadcast $trackingId (kind ${event.kind}) to ${relays.size} relays" }
|
||||
|
||||
val resultChannel = Channel<RelayResponse>(UNLIMITED)
|
||||
|
||||
val subscription =
|
||||
object : RelayConnectionListener {
|
||||
override fun onCannotConnect(
|
||||
relay: IRelayClient,
|
||||
errorMessage: String,
|
||||
) {
|
||||
if (relay.url in relays) {
|
||||
resultChannel.trySend(
|
||||
RelayResponse(
|
||||
relay = relay.url,
|
||||
result = RelayResult.Error(errorMessage),
|
||||
),
|
||||
)
|
||||
Log.d(TAG) { "[$trackingId] Cannot connect to ${relay.url}: $errorMessage" }
|
||||
}
|
||||
}
|
||||
|
||||
override fun onDisconnected(relay: IRelayClient) {
|
||||
if (relay.url in relays) {
|
||||
resultChannel.trySend(
|
||||
RelayResponse(
|
||||
relay = relay.url,
|
||||
result = RelayResult.Error("Relay disconnected before completion"),
|
||||
),
|
||||
)
|
||||
Log.d(TAG) { "[$trackingId] Disconnected from ${relay.url}" }
|
||||
}
|
||||
}
|
||||
|
||||
override fun onIncomingMessage(
|
||||
relay: IRelayClient,
|
||||
msgStr: String,
|
||||
msg: Message,
|
||||
) {
|
||||
super.onIncomingMessage(relay, msgStr, msg)
|
||||
|
||||
when (msg) {
|
||||
is OkMessage -> {
|
||||
if (msg.eventId == event.id) {
|
||||
val result =
|
||||
if (msg.success) {
|
||||
RelayResult.Success
|
||||
} else {
|
||||
RelayResult.Error(msg.message)
|
||||
}
|
||||
resultChannel.trySend(RelayResponse(relay.url, result))
|
||||
Log.d(TAG) { "[$trackingId] Response from ${relay.url}: success=${msg.success} message=${msg.message}" }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
client.addConnectionListener(subscription)
|
||||
|
||||
val finalBroadcast =
|
||||
coroutineScope {
|
||||
val resultCollector =
|
||||
async {
|
||||
val receivedRelays = mutableSetOf<NormalizedRelayUrl>()
|
||||
var currentBroadcast = broadcast
|
||||
|
||||
withTimeoutOrNull(TIMEOUT_SECONDS * 1000) {
|
||||
while (receivedRelays.size < relays.size) {
|
||||
val response = resultChannel.receive()
|
||||
|
||||
// Skip if already received (don't override success)
|
||||
if (response.relay in receivedRelays) continue
|
||||
|
||||
receivedRelays.add(response.relay)
|
||||
currentBroadcast = currentBroadcast.withResult(response.relay, response.result)
|
||||
|
||||
// Update active broadcasts with new progress
|
||||
_activeBroadcasts.update { list ->
|
||||
list.map { if (it.id == trackingId) currentBroadcast else it }.toImmutableList()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Mark remaining relays as timeout
|
||||
relays.filter { it !in receivedRelays }.forEach { relay ->
|
||||
currentBroadcast = currentBroadcast.withResult(relay, RelayResult.Timeout)
|
||||
}
|
||||
|
||||
currentBroadcast
|
||||
}
|
||||
|
||||
// Send after setting up listener
|
||||
client.publish(event, relays)
|
||||
|
||||
resultCollector.await()
|
||||
}
|
||||
|
||||
resultChannel.close()
|
||||
|
||||
// Remove from active, emit to completed
|
||||
_activeBroadcasts.update { list ->
|
||||
list.map { if (it.id == trackingId) finalBroadcast else it }.toImmutableList()
|
||||
}
|
||||
|
||||
Log.d(TAG) { "Broadcast $trackingId complete: ${finalBroadcast.successCount}/${finalBroadcast.totalRelays} success" }
|
||||
} finally {
|
||||
client.removeConnectionListener(subscription)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Marks relays as Retrying in an existing broadcast.
|
||||
* Call this before starting the retry to show immediate feedback.
|
||||
*/
|
||||
fun markRelaysRetrying(
|
||||
broadcastId: String,
|
||||
relays: Set<NormalizedRelayUrl>,
|
||||
) {
|
||||
_activeBroadcasts.update { list ->
|
||||
list
|
||||
.map { broadcast ->
|
||||
if (broadcast.id == broadcastId) {
|
||||
var updated = broadcast
|
||||
relays.forEach { relay ->
|
||||
updated = updated.withResult(relay, RelayResult.Retrying)
|
||||
}
|
||||
updated.copy(status = BroadcastStatus.IN_PROGRESS)
|
||||
} else {
|
||||
broadcast
|
||||
}
|
||||
}.toImmutableList()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Retries sending an event to failed relays using cached event.
|
||||
* Updates the existing broadcast in-place with retry results.
|
||||
*
|
||||
* @param broadcast The broadcast to retry (must be in activeBroadcasts or recently completed)
|
||||
* @param client The Nostr client
|
||||
* @param specificRelay Optional specific relay to retry (null = all failed)
|
||||
* @return Updated BroadcastEvent or null if event not in cache
|
||||
*/
|
||||
@OptIn(DelicateCoroutinesApi::class)
|
||||
suspend fun retry(
|
||||
broadcast: BroadcastEvent,
|
||||
client: INostrClient,
|
||||
specificRelay: NormalizedRelayUrl? = null,
|
||||
): BroadcastEvent {
|
||||
val event = broadcast.event
|
||||
|
||||
val relaysToRetry =
|
||||
if (specificRelay != null) {
|
||||
setOf(specificRelay)
|
||||
} else {
|
||||
broadcast.failedRelays.toSet()
|
||||
}
|
||||
|
||||
if (relaysToRetry.isEmpty()) {
|
||||
return broadcast
|
||||
}
|
||||
|
||||
// Mark relays as retrying for immediate feedback
|
||||
markRelaysRetrying(broadcast.id, relaysToRetry)
|
||||
|
||||
// If broadcast not in active list, re-add it
|
||||
if (_activeBroadcasts.value.none { it.id == broadcast.id }) {
|
||||
_activeBroadcasts.update { list ->
|
||||
var updated = broadcast
|
||||
relaysToRetry.forEach { relay ->
|
||||
updated = updated.withResult(relay, RelayResult.Retrying)
|
||||
}
|
||||
(list + updated.copy(status = BroadcastStatus.IN_PROGRESS)).toImmutableList()
|
||||
}
|
||||
}
|
||||
|
||||
// Setup result collection
|
||||
val resultChannel = Channel<RelayResponse>(UNLIMITED)
|
||||
|
||||
val subscription =
|
||||
object : RelayConnectionListener {
|
||||
override fun onCannotConnect(
|
||||
relay: IRelayClient,
|
||||
errorMessage: String,
|
||||
) {
|
||||
if (relay.url in relaysToRetry) {
|
||||
resultChannel.trySend(
|
||||
RelayResponse(
|
||||
relay = relay.url,
|
||||
result = RelayResult.Error(errorMessage),
|
||||
),
|
||||
)
|
||||
Log.d(TAG) { "[${broadcast.id}] Retry cannot connect to ${relay.url}: $errorMessage" }
|
||||
}
|
||||
}
|
||||
|
||||
override fun onDisconnected(relay: IRelayClient) {
|
||||
if (relay.url in relaysToRetry) {
|
||||
resultChannel.trySend(
|
||||
RelayResponse(
|
||||
relay = relay.url,
|
||||
result = RelayResult.Error("Relay disconnected before completion"),
|
||||
),
|
||||
)
|
||||
Log.d(TAG) { "[${broadcast.id}] Retry disconnected from ${relay.url}" }
|
||||
}
|
||||
}
|
||||
|
||||
override fun onIncomingMessage(
|
||||
relay: IRelayClient,
|
||||
msgStr: String,
|
||||
msg: Message,
|
||||
) {
|
||||
super.onIncomingMessage(relay, msgStr, msg)
|
||||
|
||||
when (msg) {
|
||||
is OkMessage -> {
|
||||
if (msg.eventId == event.id) {
|
||||
val result =
|
||||
if (msg.success) {
|
||||
RelayResult.Success
|
||||
} else {
|
||||
RelayResult.Error(msg.message)
|
||||
}
|
||||
resultChannel.trySend(RelayResponse(relay.url, result))
|
||||
Log.d(TAG) { "[${broadcast.id}] Retry response from ${relay.url}: success=${msg.success}" }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
client.addConnectionListener(subscription)
|
||||
|
||||
val finalBroadcast =
|
||||
coroutineScope {
|
||||
val resultCollector =
|
||||
async {
|
||||
val receivedRelays = mutableSetOf<NormalizedRelayUrl>()
|
||||
var currentBroadcast = _activeBroadcasts.value.find { it.id == broadcast.id } ?: broadcast
|
||||
|
||||
withTimeoutOrNull(TIMEOUT_SECONDS * 1000) {
|
||||
while (receivedRelays.size < relaysToRetry.size) {
|
||||
val response = resultChannel.receive()
|
||||
|
||||
if (response.relay !in relaysToRetry) continue
|
||||
if (response.relay in receivedRelays) continue
|
||||
|
||||
receivedRelays.add(response.relay)
|
||||
currentBroadcast = currentBroadcast.withResult(response.relay, response.result)
|
||||
|
||||
_activeBroadcasts.update { list ->
|
||||
list.map { if (it.id == broadcast.id) currentBroadcast else it }.toImmutableList()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Mark remaining as timeout
|
||||
relaysToRetry.filter { it !in receivedRelays }.forEach { relay ->
|
||||
currentBroadcast = currentBroadcast.withResult(relay, RelayResult.Timeout)
|
||||
}
|
||||
|
||||
// Recalculate status
|
||||
val newStatus =
|
||||
when {
|
||||
currentBroadcast.results.values.any { it is RelayResult.Pending || it is RelayResult.Retrying } -> {
|
||||
BroadcastStatus.IN_PROGRESS
|
||||
}
|
||||
|
||||
currentBroadcast.results.all { it.value is RelayResult.Success } -> {
|
||||
BroadcastStatus.SUCCESS
|
||||
}
|
||||
|
||||
currentBroadcast.results.none { it.value is RelayResult.Success } -> {
|
||||
BroadcastStatus.FAILED
|
||||
}
|
||||
|
||||
else -> {
|
||||
BroadcastStatus.PARTIAL
|
||||
}
|
||||
}
|
||||
currentBroadcast.copy(status = newStatus)
|
||||
}
|
||||
|
||||
client.publish(event, relaysToRetry)
|
||||
|
||||
resultCollector.await()
|
||||
}
|
||||
|
||||
client.removeConnectionListener(subscription)
|
||||
resultChannel.close()
|
||||
|
||||
// Update in active broadcasts
|
||||
_activeBroadcasts.update { list ->
|
||||
list.map { if (it.id == broadcast.id) finalBroadcast else it }.toImmutableList()
|
||||
}
|
||||
|
||||
Log.d(TAG) { "Retry complete for ${broadcast.id}: ${finalBroadcast.successCount}/${finalBroadcast.totalRelays} success" }
|
||||
|
||||
return finalBroadcast
|
||||
}
|
||||
|
||||
/**
|
||||
* Clears all active broadcasts and cache (e.g., on logout).
|
||||
*/
|
||||
fun clear() {
|
||||
_activeBroadcasts.update { persistentListOf() }
|
||||
}
|
||||
|
||||
private data class RelayResponse(
|
||||
val relay: NormalizedRelayUrl,
|
||||
val result: RelayResult,
|
||||
)
|
||||
}
|
||||
+51
@@ -0,0 +1,51 @@
|
||||
/*
|
||||
* 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.commons.util
|
||||
|
||||
import com.vitorpamplona.quartz.utils.Log
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.delay
|
||||
|
||||
suspend fun <T> retryIfException(
|
||||
debugTag: String = "RetryIfException",
|
||||
maxRetries: Int = 10,
|
||||
delayMs: Long = 1000,
|
||||
func: suspend () -> T,
|
||||
) {
|
||||
var tentative = 0
|
||||
var currentDelay = delayMs
|
||||
while (tentative < maxRetries) {
|
||||
try {
|
||||
func()
|
||||
|
||||
// if it works, finishes.
|
||||
return
|
||||
} catch (e: Exception) {
|
||||
if (e is CancellationException) throw e
|
||||
Log.e(debugTag, "Tentative $tentative failed", e)
|
||||
|
||||
delay(currentDelay)
|
||||
tentative++
|
||||
currentDelay = currentDelay * 2
|
||||
}
|
||||
}
|
||||
// gives up
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
/*
|
||||
* 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.commons.util
|
||||
|
||||
fun <T> Set<T>.togglePresenceInSet(item: T): Set<T> = if (contains(item)) minus(item) else plus(item)
|
||||
+72
@@ -0,0 +1,72 @@
|
||||
/*
|
||||
* 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.commons.preview
|
||||
|
||||
import kotlinx.cinterop.ExperimentalForeignApi
|
||||
import kotlinx.cinterop.addressOf
|
||||
import kotlinx.cinterop.usePinned
|
||||
import platform.Foundation.NSASCIIStringEncoding
|
||||
import platform.Foundation.NSData
|
||||
import platform.Foundation.NSISOLatin1StringEncoding
|
||||
import platform.Foundation.NSString
|
||||
import platform.Foundation.NSStringEncoding
|
||||
import platform.Foundation.NSUTF16BigEndianStringEncoding
|
||||
import platform.Foundation.NSUTF16LittleEndianStringEncoding
|
||||
import platform.Foundation.NSUTF32BigEndianStringEncoding
|
||||
import platform.Foundation.NSUTF32LittleEndianStringEncoding
|
||||
import platform.Foundation.NSUTF8StringEncoding
|
||||
import platform.Foundation.NSWindowsCP1252StringEncoding
|
||||
import platform.Foundation.create
|
||||
|
||||
/**
|
||||
* iOS decode of HTML bytes by charset name. The common web charsets are mapped
|
||||
* to their `NSStringEncoding`; anything else falls back to UTF-8 (matching the
|
||||
* "defaults to UTF-8" behaviour of the charset sniffer).
|
||||
*/
|
||||
@OptIn(ExperimentalForeignApi::class)
|
||||
actual fun decodeBytes(
|
||||
bytes: ByteArray,
|
||||
charsetName: String?,
|
||||
): String {
|
||||
if (bytes.isEmpty()) return ""
|
||||
|
||||
val encoding = encodingFor(charsetName)
|
||||
|
||||
val data =
|
||||
bytes.usePinned { pinned ->
|
||||
NSData.create(bytes = pinned.addressOf(0), length = bytes.size.toULong())
|
||||
}
|
||||
|
||||
return (NSString.create(data, encoding) as String?)
|
||||
?: bytes.decodeToString()
|
||||
}
|
||||
|
||||
private fun encodingFor(charsetName: String?): NSStringEncoding =
|
||||
when (charsetName?.trim()?.uppercase()) {
|
||||
"UTF-16", "UTF-16BE", "UTF16" -> NSUTF16BigEndianStringEncoding
|
||||
"UTF-16LE" -> NSUTF16LittleEndianStringEncoding
|
||||
"UTF-32", "UTF-32BE", "UTF32" -> NSUTF32BigEndianStringEncoding
|
||||
"UTF-32LE" -> NSUTF32LittleEndianStringEncoding
|
||||
"ISO-8859-1", "ISO8859-1", "ISO_8859-1", "LATIN1", "L1", "CP819" -> NSISOLatin1StringEncoding
|
||||
"WINDOWS-1252", "CP1252" -> NSWindowsCP1252StringEncoding
|
||||
"US-ASCII", "ASCII", "ANSI_X3.4-1968" -> NSASCIIStringEncoding
|
||||
else -> NSUTF8StringEncoding
|
||||
}
|
||||
+34
@@ -0,0 +1,34 @@
|
||||
/*
|
||||
* 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.commons.preview
|
||||
|
||||
import java.nio.charset.Charset
|
||||
|
||||
actual fun decodeBytes(
|
||||
bytes: ByteArray,
|
||||
charsetName: String?,
|
||||
): String {
|
||||
val charset =
|
||||
charsetName
|
||||
?.let { runCatching { Charset.forName(it) }.getOrNull() }
|
||||
?: Charsets.UTF_8
|
||||
return bytes.toString(charset)
|
||||
}
|
||||
+93
@@ -0,0 +1,93 @@
|
||||
/*
|
||||
* 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.commons.preview
|
||||
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.withContext
|
||||
import okhttp3.MediaType.Companion.toMediaType
|
||||
import okhttp3.OkHttpClient
|
||||
import okhttp3.Request
|
||||
import okhttp3.coroutines.executeAsync
|
||||
|
||||
class UrlPreview {
|
||||
suspend fun fetch(
|
||||
url: String,
|
||||
okHttpClient: (String) -> OkHttpClient,
|
||||
onComplete: suspend (urlInfo: UrlInfoItem) -> Unit,
|
||||
onFailed: suspend (t: Throwable) -> Unit,
|
||||
) = try {
|
||||
onComplete(getDocument(url, okHttpClient))
|
||||
} catch (t: Throwable) {
|
||||
if (t is CancellationException) throw t
|
||||
onFailed(t)
|
||||
}
|
||||
|
||||
suspend fun getDocument(
|
||||
url: String,
|
||||
okHttpClient: (String) -> OkHttpClient,
|
||||
): UrlInfoItem =
|
||||
withContext(Dispatchers.IO) {
|
||||
val request =
|
||||
Request
|
||||
.Builder()
|
||||
.url(url)
|
||||
.get()
|
||||
.build()
|
||||
|
||||
val client = okHttpClient(url)
|
||||
|
||||
client.newCall(request).executeAsync().use { response ->
|
||||
withContext(Dispatchers.IO) {
|
||||
if (response.isSuccessful) {
|
||||
val mimeType =
|
||||
response.headers["Content-Type"]?.toMediaType()
|
||||
?: throw IllegalArgumentException("Website returned unknown mimetype: ${response.headers["Content-Type"]}")
|
||||
when {
|
||||
mimeType.type == "text" && mimeType.subtype == "html" -> {
|
||||
val metaTags = HtmlParser().parseHtml(response.body.bytes(), mimeType.charset()?.name())
|
||||
val data = OpenGraphParser().extractUrlInfo(metaTags)
|
||||
UrlInfoItem(url, data.title, data.description, data.image, mimeType.toString())
|
||||
}
|
||||
|
||||
mimeType.type == "image" -> {
|
||||
UrlInfoItem(url, image = url, mimeType = mimeType.toString())
|
||||
}
|
||||
|
||||
mimeType.type == "video" -> {
|
||||
UrlInfoItem(url, image = url, mimeType = mimeType.toString())
|
||||
}
|
||||
|
||||
mimeType.type == "application" && mimeType.subtype == "pdf" -> {
|
||||
UrlInfoItem(url, image = url, mimeType = mimeType.toString())
|
||||
}
|
||||
|
||||
else -> {
|
||||
throw IllegalArgumentException("Website returned unknown encoding for previews: $mimeType")
|
||||
}
|
||||
}
|
||||
} else {
|
||||
throw IllegalArgumentException("Website returned: " + response.code)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user