Implement subscriptions via the graphql websocket protocol

androidwalletdemo/src/main/java/com/lightspark/androidwalletdemo/auth/CredentialsStore.kt

@@ -46,7 +46,7 @@ class CredentialsStore(private val context: Context) {
     fun getJwtInfoFlow() = context.dataStore.data.map { preferences ->
         val accountId = preferences[ACCOUNT_ID_KEY] ?: return@map null
         val jwt = preferences[JWT_KEY] ?: return@map null
-        val userName = preferences[USER_NAME_KEY] ?: return@map null
+        val userName = preferences[USER_NAME_KEY] ?: ""
         SavedCredentials(accountId, jwt, userName)
     }.distinctUntilChanged()
 }

androidwalletdemo/src/main/java/com/lightspark/androidwalletdemo/sendpayment/SendPaymentViewModel.kt

@@ -4,10 +4,12 @@ import android.util.Log
 import androidx.lifecycle.ViewModel
 import androidx.lifecycle.viewModelScope
 import com.lightspark.androidwalletdemo.util.CurrencyAmountArg
+import com.lightspark.androidwalletdemo.util.currencyAmountSats
 import com.lightspark.androidwalletdemo.util.zeroCurrencyAmount
 import com.lightspark.androidwalletdemo.util.zeroCurrencyAmountArg
 import com.lightspark.androidwalletdemo.wallet.PaymentRepository
 import com.lightspark.sdk.core.Lce
+import com.lightspark.sdk.wallet.model.TransactionStatus
 import dagger.hilt.android.lifecycle.HiltViewModel
 import javax.inject.Inject
 import kotlinx.coroutines.ExperimentalCoroutinesApi
@@ -42,7 +44,16 @@ class SendPaymentViewModel @Inject constructor(
             }
         }.filterNotNull().map {
             when (it) {
-                is Lce.Content -> PaymentStatus.SUCCESS
+                is Lce.Content -> {
+                    when (it.data.status) {
+                        TransactionStatus.PENDING -> PaymentStatus.PENDING
+                        TransactionStatus.SUCCESS -> PaymentStatus.SUCCESS
+                        else -> {
+                            Log.e("SendPaymentViewModel", "Error sending payment")
+                            PaymentStatus.FAILURE
+                        }
+                    }
+                }
                 is Lce.Error -> {
                     Log.e("SendPaymentViewModel", "Error sending payment", it.exception)
                     PaymentStatus.FAILURE
@@ -62,7 +73,9 @@ class SendPaymentViewModel @Inject constructor(
         .onEach {
             (it as? Lce.Content)?.let { invoiceData ->
                 invoiceData.data?.amount?.let { decodedInvoiceAmount ->
-                    paymentAmountFlow.tryEmit(decodedInvoiceAmount)
+                    // Default to paying 10 sats for 0 amount invoices.
+                    val amount = if (decodedInvoiceAmount.originalValue > 0) decodedInvoiceAmount else currencyAmountSats(10)
+                    paymentAmountFlow.tryEmit(amount)
                 }
             }
         }

androidwalletdemo/src/main/java/com/lightspark/androidwalletdemo/wallet/PaymentRepository.kt

@@ -1,6 +1,7 @@
 package com.lightspark.androidwalletdemo.wallet
 
 import com.lightspark.androidwalletdemo.util.CurrencyAmountArg
+import com.lightspark.sdk.core.asLce
 import com.lightspark.sdk.core.wrapWithLceFlow
 import com.lightspark.sdk.wallet.LightsparkCoroutinesWalletClient
 import com.lightspark.sdk.wallet.model.InvoiceType
@@ -16,8 +17,8 @@ class PaymentRepository @Inject constructor(private val lightsparkClient: Lights
             lightsparkClient.createInvoice(amountMillis, memo, type)
         }.flowOn(Dispatchers.IO)
 
-    fun payInvoice(invoice: String) =
-        wrapWithLceFlow { lightsparkClient.payInvoice(invoice, 1000000) }.flowOn(Dispatchers.IO)
+    suspend fun payInvoice(invoice: String) =
+        lightsparkClient.payInvoiceAndAwaitCompletion(invoice, 1000000).asLce().flowOn(Dispatchers.IO)
 
     fun decodeInvoice(encodedInvoice: String) =
         wrapWithLceFlow { lightsparkClient.decodeInvoice(encodedInvoice) }.flowOn(Dispatchers.IO)

androidwalletdemo/src/main/java/com/lightspark/androidwalletdemo/wallet/WalletRepository.kt

@@ -6,6 +6,7 @@ import com.lightspark.sdk.core.Lce
 import com.lightspark.sdk.core.asLce
 import com.lightspark.sdk.core.auth.AuthProvider
 import com.lightspark.sdk.core.crypto.androidKeystoreContainsPrivateKeyForAlias
+import com.lightspark.sdk.core.crypto.generateSigningKeyPair
 import com.lightspark.sdk.core.crypto.generateSigningKeyPairInAndroidKeyStore
 import com.lightspark.sdk.core.requester.ServerEnvironment
 import com.lightspark.sdk.core.wrapWithLceFlow
@@ -51,8 +52,8 @@ class WalletRepository @Inject constructor(
         // able to export that key for the user, but it does come with increased security. If you'd like to manage your
         // own keys or store them in some other way in your own app code, you can still generate a valid key pair using
         // the [generateSigningKeyPair] function in the SDK.
-        val keyPair = generateSigningKeyPairInAndroidKeyStore(LIGHTSPARK_SIGNING_KEY_ALIAS)
-        walletClient.loadWalletSigningKeyAlias(LIGHTSPARK_SIGNING_KEY_ALIAS)
+        val keyPair = generateSigningKeyPair()
+        walletClient.loadWalletSigningKey(keyPair.private.encoded)
         return walletClient.initializeWalletAndWaitForInitialized(
             keyType = KeyType.RSA_OAEP,
             signingPublicKey = Base64.encodeToString(keyPair.public.encoded, Base64.NO_WRAP),

core/src/commonMain/kotlin/com/lightspark/sdk/core/requester/GraphQLWebsocketProtocol.kt

@@ -0,0 +1,185 @@
+package com.lightspark.sdk.core.requester
+
+import com.lightspark.sdk.core.LightsparkErrorCode
+import com.lightspark.sdk.core.LightsparkException
+import io.ktor.client.plugins.websocket.ClientWebSocketSession
+import io.ktor.websocket.close
+import io.ktor.websocket.send
+import kotlinx.coroutines.flow.catch
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.withTimeout
+import kotlinx.serialization.decodeFromString
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.jsonObject
+import kotlinx.serialization.json.jsonPrimitive
+
+const val CONNECTION_INIT_TIMEOUT = 10_000L
+
+/**
+ * An implementation of https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md for use with a Ktor client.
+ * It can carry queries in addition to subscriptions over the websocket
+ */
+internal class GraphQLWebsocketProtocol(
+    private val webSocketSession: ClientWebSocketSession,
+    private val listener: GraphQLWebsocketListener,
+    private val jsonSerialFormat: Json,
+    private val connectionPayload: suspend () -> JsonObject? = { null },
+) {
+    suspend fun connectionInit() {
+        val payload = connectionPayload()
+
+        sendMessage {
+            add("type", "connection_init")
+            payload?.let { add("payload", it) }
+        }
+        waitForConnectionAck()
+    }
+
+    suspend fun run() {
+        // TODO(Jeremy): Consider adding client-side ping.
+        webSocketSession.incoming
+            .receiveAsFlow()
+            .catch { cause ->
+                listener.onError(cause)
+            }
+            .map {
+                jsonSerialFormat.decodeFromString(JsonObject.serializer(), it.data.decodeToString())
+            }.collect {
+                handleServerMessage(it)
+            }
+    }
+
+    suspend fun <T> sendQuery(id: String, query: Query<T>) {
+        val operationNameRegex =
+            Regex("^\\s*(query|mutation|subscription)\\s+(\\w+)", RegexOption.IGNORE_CASE)
+        val operationMatch = operationNameRegex.find(query.queryPayload)
+        if (operationMatch == null || operationMatch.groupValues.size < 3) {
+            throw LightsparkException("Invalid query payload", LightsparkErrorCode.INVALID_QUERY)
+        }
+        val operation = operationMatch.groupValues[2]
+        // TODO(Jeremy): Handle the signing node ID.
+        sendMessage {
+            add("id", id)
+            add("type", "subscribe")
+            val payload = buildJsonObject(jsonSerialFormat) {
+                add("query", query.queryPayload)
+                add("variables", buildJsonObject(jsonSerialFormat, query.variableBuilder))
+                add("operationName", operation)
+            }
+            add("payload", payload)
+        }
+    }
+
+    suspend fun stopQuery(id: String) {
+        sendMessage {
+            add("id", id)
+            add("type", "complete")
+        }
+    }
+
+    suspend fun close() {
+        webSocketSession.close()
+    }
+
+    private suspend fun waitForConnectionAck() {
+        withTimeout(CONNECTION_INIT_TIMEOUT) {
+            while (true) {
+                val received = webSocketSession.incoming.receive()
+                try {
+                    val receivedText = received.data.decodeToString()
+                    val receivedJson = jsonSerialFormat.decodeFromString<JsonObject>(
+                        receivedText,
+                    )
+                    when (val type = receivedJson["type"]?.jsonPrimitive?.content) {
+                        "connection_ack" -> return@withTimeout
+                        "ping" -> sendPong()
+                        else -> {
+                            listener.onError(Exception("Unexpected message type: $type"))
+                            return@withTimeout
+                        }
+                    }
+                } catch (e: Exception) {
+                    listener.onError(e)
+                    return@withTimeout
+                }
+            }
+        }
+    }
+
+    private suspend fun handleServerMessage(messageJson: JsonObject) {
+        when (messageJson["type"]?.jsonPrimitive?.content) {
+            "next" -> {
+                val payload = messageJson["payload"]?.jsonObject
+                val id = messageJson["id"]?.jsonPrimitive?.content
+                if (id != null && payload != null) {
+                    listener.onOperationMessage(id, payload)
+                }
+            }
+
+            "error" -> {
+                val payload = messageJson["payload"]?.jsonObject
+                val id = messageJson["id"]?.jsonPrimitive?.content
+                val errors = payload?.get("errors")?.jsonObject
+                if (id != null && errors != null) {
+                    listener.onOperationError(id, errors)
+                }
+            }
+
+            "complete" -> {
+                messageJson["id"]?.jsonPrimitive?.content?.let {
+                    listener.operationComplete(it)
+                }
+            }
+
+            "ping" -> sendPong()
+            "pong" -> Unit // Nothing to do, the server acknowledged one of our pings
+            else -> Unit // Unknown message
+        }
+    }
+
+    private suspend fun sendPong() {
+        sendMessage {
+            add("type", "pong")
+        }
+    }
+
+    private suspend fun sendMessage(payloadBuilder: JsonObjectBuilder.() -> Unit) {
+        val jsonObjectBuilder = buildJsonObject(jsonSerialFormat, payloadBuilder)
+        webSocketSession.send(jsonObjectBuilder.toString())
+    }
+}
+
+internal interface GraphQLWebsocketListener {
+    fun onOperationMessage(id: String, payload: JsonObject)
+    fun onOperationError(id: String, payload: JsonObject)
+    fun operationComplete(id: String)
+    fun onError(error: Throwable)
+    fun onClose(code: Int, reason: String)
+}
+
+@Suppress("unused")
+enum class CloseCode(val code: Int) {
+    NormalClosure(1000),
+    GoingAway(1001),
+    AbnormalClosure(1006),
+    NoStatusReceived(1005),
+    ServiceRestart(1012),
+    TryAgainLater(1013),
+    BadGateway(1013),
+
+    InternalServerError(4500),
+    InternalClientError(4005),
+    BadRequest(4400),
+    BadResponse(4004),
+    Unauthorized(4401),
+    Forbidden(4403),
+    SubprotocolNotAcceptable(4406),
+    ConnectionInitialisationTimeout(4408),
+    ConnectionAcknowledgementTimeout(4504),
+    SubscriberAlreadyExists(4409),
+    TooManyInitialisationRequests(4429),
+
+    Terminated(4499),
+}

core/src/commonMain/kotlin/com/lightspark/sdk/core/requester/VariableBuilder.kt → core/src/commonMain/kotlin/com/lightspark/sdk/core/requester/JsonObjectBuilder.kt

@@ -6,7 +6,7 @@ import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.json.JsonPrimitive
 import kotlinx.serialization.json.encodeToJsonElement
 
-class VariableBuilder(val jsonSerialFormat: Json) {
+class JsonObjectBuilder(val jsonSerialFormat: Json) {
     val variables = mutableMapOf<String, JsonElement>()
 
     fun add(name: String, value: JsonElement) {
@@ -38,8 +38,16 @@ class VariableBuilder(val jsonSerialFormat: Json) {
     }
 }
 
-fun variables(jsonSerialFormat: Json, builder: VariableBuilder.() -> Unit): JsonObject {
-    val variableBuilder = VariableBuilder(jsonSerialFormat)
-    variableBuilder.builder()
-    return variableBuilder.build()
+fun buildJsonObject(jsonSerialFormat: Json, builder: JsonObjectBuilder.() -> Unit): JsonObject {
+    val jsonObjectBuilder = JsonObjectBuilder(jsonSerialFormat)
+    jsonObjectBuilder.builder()
+    return jsonObjectBuilder.build()
+}
+
+fun Map<String, Any?>.toJsonObject(jsonSerialFormat: Json): JsonObject {
+    val jsonObjectBuilder = JsonObjectBuilder(jsonSerialFormat)
+    forEach { (key, value) ->
+        jsonObjectBuilder.add(key, value)
+    }
+    return jsonObjectBuilder.build()
 }

core/src/commonMain/kotlin/com/lightspark/sdk/core/requester/Query.kt

@@ -1,5 +1,6 @@
 package com.lightspark.sdk.core.requester
 
+import java.util.UUID
 import kotlin.jvm.JvmOverloads
 import kotlinx.serialization.json.JsonObject
 
@@ -9,10 +10,12 @@ interface StringDeserializer<T> {
 
 data class Query<T>(
     val queryPayload: String,
-    val variableBuilder: VariableBuilder.() -> Unit,
+    val variableBuilder: JsonObjectBuilder.() -> Unit,
     val signingNodeId: String? = null,
     val deserializer: (JsonObject) -> T,
 ) {
+    val id = UUID.randomUUID().toString()
+
     /**
      * This constructor is for convenience when calling from Java rather than Kotlin. The primary constructor is
      * simpler to use from Kotlin if possible.