feat: support user-scoped execution SSE endpoint

Author: SIMULATAN

api/src/main/kotlin/me/snoty/backend/utils/SerializationUtils.kt

@@ -5,5 +5,9 @@ import kotlinx.serialization.json.Json
 import kotlinx.serialization.serializer
 import kotlin.reflect.KClass
 
+/**
+ * A hacky way to encode an object to a JSON string using the serializer of its class.
+ * This allows users to serialize objects whose classes are not known at compile time.
+ */
 @OptIn(InternalSerializationApi::class)
 fun <T : Any> Json.hackyEncodeToString(it: T) = encodeToString((it::class as KClass<T>).serializer(), it)

api/src/main/kotlin/me/snoty/backend/wiring/flow/execution/FlowExecutionEventService.kt

@@ -7,28 +7,33 @@ import me.snoty.backend.integration.config.flow.NodeId
 import me.snoty.backend.scheduling.FlowTriggerReason
 import me.snoty.integration.common.wiring.flow.FlowExecutionStatus
 import me.snoty.integration.common.wiring.flow.NodeLogEntry
+import kotlin.uuid.Uuid
 
 @Serializable
-sealed class FlowExecutionEvent(@Transient val eventType: String) {
+sealed class FlowExecutionEvent(val eventType: String) {
+	abstract val userId: Uuid
 	abstract val flowId: NodeId
 	val timestamp = Clock.System.now()
 
 	@Serializable
 	data class FlowStartedEvent(
+		override val userId: Uuid,
 		override val flowId: String,
 		val jobId: String,
 		val triggeredBy: FlowTriggerReason,
 	) : FlowExecutionEvent("FlowStarted")
 
 	@Serializable
 	data class FlowLogEvent(
+		override val userId: Uuid,
 		override val flowId: String,
 		val jobId: String,
 		val entry: NodeLogEntry,
 	) : FlowExecutionEvent("FlowLog")
 
 	@Serializable
 	data class FlowEndedEvent(
+		override val userId: Uuid,
 		override val flowId: String,
 		val jobId: String,
 		val status: FlowExecutionStatus,
@@ -42,4 +47,4 @@ interface FlowExecutionEventService {
 	 * Offers an event. Will be passed on using the database or in-memory channel.
 	 */
 	suspend fun offer(event: FlowExecutionEvent)
-}
+}

src/main/kotlin/me/snoty/backend/integration/flow/execution/FlowRunnerImpl.kt

@@ -51,6 +51,7 @@ class FlowRunnerImpl(
 		flowExecutionService.create(jobId, flow._id, triggeredBy)
 
 		flowExecutionEventService.offer(FlowExecutionEvent.FlowStartedEvent(
+			userId = flow.userId,
 			flowId = flow._id,
 			jobId = jobId,
 			triggeredBy = triggeredBy,
@@ -78,6 +79,7 @@ class FlowRunnerImpl(
 				rootSpan.end()
 				val status = if (it == null) FlowExecutionStatus.SUCCESS else FlowExecutionStatus.FAILED
 				flowExecutionEventService.offer(FlowExecutionEvent.FlowEndedEvent(
+					userId = flow.userId,
 					flowId = flow._id,
 					jobId = jobId,
 					status = status,

src/main/kotlin/me/snoty/backend/integration/flow/execution/FlowTracing.kt

@@ -32,9 +32,11 @@ class FlowTracingImpl(
 		val flowId = flow._id
 
 		val rootSpan = spanBuilder("Flow $flowId")
+			.setAttribute(USER_ID, flow.userId)
 			.setAttribute(JOB_ID, jobId)
 			.setAttribute(FLOW_ID, flowId)
 			.startSpan()
+		KMDC.put(USER_ID, flow._id)
 
 		return rootSpan
 	}
@@ -43,8 +45,6 @@ class FlowTracingImpl(
 		setAttribute(NODE_ID, node._id)
 		KMDC.put(NODE_ID, node._id)
 		setAttribute("node.descriptor", node.descriptor)
-		setAttribute(USER_ID, node.userId)
-		KMDC.put(USER_ID, node.userId.toString())
 
 		if (featureFlags.traceConfig) {
 			setAttribute("config", json.encodeToString(node.settings))

src/main/kotlin/me/snoty/backend/integration/flow/logging/NodeLogAppender.kt

@@ -9,10 +9,8 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.newSingleThreadContext
 import kotlinx.datetime.Instant
 import me.snoty.backend.logging.toSLF4JLevel
-import me.snoty.backend.observability.APPENDER_LOG_LEVEL
-import me.snoty.backend.observability.FLOW_ID
-import me.snoty.backend.observability.JOB_ID
-import me.snoty.backend.observability.NODE_ID
+import me.snoty.backend.observability.*
+import me.snoty.backend.utils.toUuid
 import me.snoty.backend.wiring.flow.execution.FlowExecutionEvent
 import me.snoty.backend.wiring.flow.execution.FlowExecutionEventService
 import me.snoty.backend.wiring.flow.execution.FlowExecutionService
@@ -60,7 +58,8 @@ class NodeLogAppender(
 		scope.launch {
 			flowExecutionEventService.offer(FlowExecutionEvent.FlowLogEvent(
 				jobId = jobId,
-				flowId = eventObject.mdcPropertyMap[FLOW_ID.key] ?: return@launch logger.warn { "No flow ID found in log entry with msg='$message'" },
+				userId = eventObject.mdcPropertyMap[USER_ID.key]?.toUuid() ?: return@launch logger.error { "No User ID found in log entry with msg='$message'" },
+				flowId = eventObject.mdcPropertyMap[FLOW_ID.key] ?: return@launch logger.error { "No Flow ID found in log entry with msg='$message'" },
 				entry = entry,
 			))
 

src/main/kotlin/me/snoty/backend/scheduling/jobrunr/node/JobRunrFlowJobHandler.kt

@@ -2,6 +2,7 @@ package me.snoty.backend.scheduling.jobrunr.node
 
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.slf4j.MDCContext
+import kotlinx.coroutines.withContext
 import me.snoty.backend.integration.flow.logging.NodeLogAppender
 import me.snoty.backend.logging.KMDC
 import me.snoty.backend.observability.APPENDER_LOG_LEVEL
@@ -50,16 +51,18 @@ class JobRunrFlowJobHandler(
 
 			KMDC.put(USER_ID, flow.userId.toString())
 
-			logger.debug("Processing flow {}", flow)
+			withContext(MDCContext()) {
+				logger.debug("Processing flow {}", flow)
 
-			flowRunner.execute(
-				jobId = jobContext.jobId.toString(),
-				triggeredBy = jobRequest.triggeredBy,
-				logger = logger,
-				logLevel = jobRequest.logLevel,
-				flow = flow,
-				input = SimpleIntermediateData(jobContext),
-			)
+				flowRunner.execute(
+					jobId = jobContext.jobId.toString(),
+					triggeredBy = jobRequest.triggeredBy,
+					logger = logger,
+					logLevel = jobRequest.logLevel,
+					flow = flow,
+					input = SimpleIntermediateData(jobContext),
+				)
+			}
 		}
 	}
 }

src/main/kotlin/me/snoty/backend/server/resources/wiring/flow/FlowExecutionResource.kt

@@ -6,43 +6,63 @@ import io.ktor.server.sse.*
 import io.ktor.sse.*
 import kotlinx.coroutines.flow.filter
 import kotlinx.serialization.json.Json
+import me.snoty.backend.utils.getUser
+import me.snoty.backend.utils.hackyEncodeToString
 import me.snoty.backend.utils.orNull
 import me.snoty.backend.wiring.flow.execution.FlowExecutionEventService
 import me.snoty.backend.wiring.flow.execution.FlowExecutionService
 import org.koin.ktor.ext.get
 import kotlin.time.Duration.Companion.seconds
 
-fun Route.flowExecutionResource() = route("{id}") {
+fun Route.flowExecutionResource() {
 	val flowExecutionService: FlowExecutionService = get()
+	val flowExecutionEventService: FlowExecutionEventService = get()
+	val json: Json = get()
+
+	route("{id}") {
+		get("executions") {
+			val startFrom = call.request.queryParameters["startFrom"]?.orNull()
+			val limit = call.request.queryParameters["limit"]?.toIntOrNull() ?: 10
+			val flow = getPersonalFlowOrNull() ?: return@get
+			val executions = flowExecutionService.query(flowId = flow._id, startFrom = startFrom, limit = limit)
 
-	get("executions") {
-		val startFrom = call.request.queryParameters["startFrom"]?.orNull()
-		val limit = call.request.queryParameters["limit"]?.toIntOrNull() ?: 10
-		val flow = getPersonalFlowOrNull() ?: return@get
-		val executions = flowExecutionService.query(flowId = flow._id, startFrom = startFrom, limit = limit)
+			call.respond(executions)
+		}
+
+		sse("executions/sse") {
+			val flow = getPersonalFlowOrNull() ?: return@sse
 
-		call.respond(executions)
+			heartbeat {
+				period = 10.seconds
+				event = ServerSentEvent("heartbeat")
+			}
+			val eventTypes = call.request.queryParameters["eventTypes"]?.split(",")?.map { it.trim() } ?: emptyList()
+
+			flowExecutionEventService.provideBus()
+				.filter { it.flowId == flow._id }
+				.filter { eventTypes.isEmpty() || it.eventType in eventTypes }
+				.collect {
+					send(
+						data = json.encodeToString(it),
+						event = it.eventType,
+					)
+				}
+		}
 	}
 
-	val flowExecutionEventService: FlowExecutionEventService = get()
-	val json: Json = get()
-	
 	sse("executions/sse") {
-		val flow = getPersonalFlowOrNull() ?: return@sse
+		val user = call.getUser()
 
 		heartbeat {
 			period = 10.seconds
 			event = ServerSentEvent("heartbeat")
 		}
 
-		val eventTypes = call.request.queryParameters["eventTypes"]?.split(",")?.map { it.trim() } ?: emptyList()
-
 		flowExecutionEventService.provideBus()
-			.filter { it.flowId == flow._id }
-			.filter { eventTypes.isEmpty() || it.eventType in eventTypes }
+			.filter { it.userId == user.id }
 			.collect {
 				send(
-					data = json.encodeToString(it),
+					data = json.hackyEncodeToString(it),
 					event = it.eventType,
 				)
 			}