feat(node): support receiving empty inputs

Useful for the Diff Injector Node, which will react to elements removed from the input.

Author: SIMULATAN

api/src/main/kotlin/me/snoty/integration/common/annotation/RegisterNode.kt

@@ -24,3 +24,8 @@ annotation class RegisterNode(
 	val inputType: KClass<out Any> = NoSchema::class,
 	val outputType: KClass<out Any> = NoSchema::class,
 )
+
+/**
+ * Marks that the node wants to receive empty input. Otherwise, following nodes will not be executed at all if no input element was produced by the prior node.
+ */
+annotation class ReceiveEmptyInput

api/src/main/kotlin/me/snoty/integration/common/model/metadata/NodeMetadata.kt

@@ -18,6 +18,7 @@ data class NodeMetadata(
 	@Transient
 	val settingsClass: KClass<out NodeSettings> = NodeSettings::class, // previously threw an error, but since we deserialize this AND remain compatible, we need to keep this
 	val input: ObjectSchema?,
+	val receiveEmptyInput: Boolean = false,
 	val output: ObjectSchema?
 )
 

integration-plugin/src/main/kotlin/me/snoty/integration/plugin/utils/MetadataGenerator.kt

@@ -4,6 +4,7 @@ import com.google.devtools.ksp.KspExperimental
 import com.google.devtools.ksp.processing.Resolver
 import com.google.devtools.ksp.symbol.KSClassDeclaration
 import me.snoty.backend.wiring.node.metadataJson
+import me.snoty.integration.common.annotation.ReceiveEmptyInput
 import me.snoty.integration.common.annotation.RegisterNode
 import me.snoty.integration.common.model.metadata.NodeMetadata
 import me.snoty.integration.common.wiring.node.NodeSettings
@@ -20,7 +21,8 @@ fun generateMetadata(resolver: Resolver, clazz: KSClassDeclaration, node: Regist
 		settingsClass = NodeSettings::class,
 		settings = generateObjectSchema(resolver, settingsClass)!!,
 		input = generateObjectSchema(resolver, inputClass),
-		output = generateObjectSchema(resolver, outputClass)
+		receiveEmptyInput = clazz.hasAnnotation<ReceiveEmptyInput>(),
+		output = generateObjectSchema(resolver, outputClass),
 	)
 
 	return metadataJson.encodeToString(metadata)

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

@@ -192,8 +192,6 @@ class FlowRunnerImpl(
 			}
 			.flowOn(span.asContextElement() + MDCContext())
 			.flatMapConcat { output ->
-				@Suppress("UnusedFlow") // yeah I don't think so
-				if (output.isEmpty()) return@flatMapConcat emptyFlow()
 				node.next
 					.asFlow()
 					.mapNotNull { nextNodeId ->
@@ -202,6 +200,7 @@ class FlowRunnerImpl(
 							null
 						}
 					}
+					.filter { nextNode -> !output.isEmpty() || nodeRegistry.getMetadata(nextNode.descriptor).receiveEmptyInput }
 					.flatMapConcat { nextNode ->
 						val subspan = span.subspan(flowTracing, traceName(nextNode)) {
 							setNodeAttributes(nextNode, output)