[wip] migrate legacy integrations to node-based system

Author: SIMULATAN

api/src/main/kotlin/me/snoty/backend/database/mongo/BsonUtils.kt

@@ -9,6 +9,7 @@ import org.bson.codecs.DecoderContext
 import org.bson.codecs.DocumentCodec
 import org.bson.codecs.EncoderContext
 import org.bson.codecs.configuration.CodecRegistry
+import kotlin.reflect.KClass
 
 // the joys of mongodb
 fun <T : Any> CodecRegistry.encode(value: T): Document {
@@ -22,3 +23,14 @@ fun <T : Any> CodecRegistry.encode(value: T): Document {
 
 	return documentCodec.decode(BsonDocumentReader(document), DecoderContext.builder().build())
 }
+
+
+// must be inlined to correctly create the context
+@Suppress("NOTHING_TO_INLINE")
+inline fun <T : Any> CodecRegistry.decode(clazz: KClass<T>, document: Document): T {
+	return get(clazz.java)
+		.decode(
+			BsonDocumentReader(document.toBsonDocument()),
+			DecoderContext.builder().build()
+		)
+}

api/src/main/kotlin/me/snoty/backend/errors/ServiceResult.kt

@@ -0,0 +1,11 @@
+package me.snoty.backend.errors
+
+import io.ktor.http.*
+
+open class ServiceResult(
+	val httpCode: Int,
+	val message: String,
+	val details: String? = null
+) {
+	constructor(httpCode: HttpStatusCode, message: String, details: String? = null) : this(httpCode.value, message, details)
+}

api/src/main/kotlin/me/snoty/backend/integration/config/ConfigId.kt

@@ -1,5 +1,23 @@
 package me.snoty.backend.integration.config.flow
 
+import kotlinx.serialization.KSerializer
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.descriptors.SerialDescriptor
+import kotlinx.serialization.descriptors.serialDescriptor
+import kotlinx.serialization.encoding.Decoder
+import kotlinx.serialization.encoding.Encoder
 import org.bson.types.ObjectId
 
-typealias NodeId = ObjectId
+typealias NodeId = @Serializable(NodeIdSerializer::class) ObjectId
+
+object NodeIdSerializer : KSerializer<NodeId> {
+	override val descriptor: SerialDescriptor = serialDescriptor<String>()
+
+	override fun deserialize(decoder: Decoder): NodeId {
+		return NodeId(decoder.decodeString())
+	}
+
+	override fun serialize(encoder: Encoder, value: NodeId) {
+		encoder.encodeString(value.toString())
+	}
+}

api/src/main/kotlin/me/snoty/backend/integration/config/flow/NodeId.kt

@@ -1,3 +1,17 @@
 package me.snoty.backend.utils
 
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.transform
+
 fun <T, C : Collection<T>> C.orNull() = ifEmpty { null }
+
+inline fun <T> T.contextual(block: T.() -> Unit) {
+	block()
+}
+
+fun <T, R> Flow<T>.listAsElements(block: suspend (T) -> Collection<R>): Flow<R> = this.transform { item ->
+	val result = block(item)
+	result.forEach {
+		emit(it)
+	}
+}

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

@@ -32,6 +32,7 @@ allprojects {
         kotlin {
             compilerOptions {
                 freeCompilerArgs.add("-Xcontext-receivers")
+                optIn.add("kotlinx.coroutines.ExperimentalCoroutinesApi")
             }
         }
     }

build.gradle.kts

@@ -1,13 +1,17 @@
 package me.snoty.integration.common
 
 import me.snoty.backend.scheduling.Scheduler
-import me.snoty.integration.common.config.IntegrationConfigService
+import me.snoty.integration.common.config.NodeService
 import me.snoty.integration.common.diff.EntityStateService
+import me.snoty.integration.common.wiring.flow.FlowService
 import me.snoty.integration.common.utils.calendar.CalendarService
+import org.bson.codecs.configuration.CodecRegistry
 
-data class IntegrationContext(
+data class NodeContext(
 	val entityStateService: EntityStateService,
-	val integrationConfigService: IntegrationConfigService,
+	val nodeService: NodeService,
+	val flowService: FlowService,
+	val codecRegistry: CodecRegistry,
 	val calendarService: CalendarService,
 	val scheduler: Scheduler
 )

integrations/api/src/main/kotlin/me/snoty/integration/common/AbstractIntegration.kt

@@ -0,0 +1,34 @@
+package me.snoty.integration.common.config
+
+import io.ktor.http.*
+import kotlinx.coroutines.flow.Flow
+import me.snoty.backend.errors.ServiceResult
+import me.snoty.backend.integration.config.flow.NodeId
+import me.snoty.integration.common.wiring.IFlowNode
+import me.snoty.integration.common.wiring.node.NodeDescriptor
+import me.snoty.integration.common.wiring.node.NodePosition
+import me.snoty.integration.common.wiring.node.NodeSettings
+import java.util.*
+
+interface NodeService {
+	fun getByUser(userID: UUID, position: NodePosition?): Flow<IFlowNode>
+
+	fun getAll(integrationType: String): Flow<IFlowNode>
+
+	suspend fun get(id: NodeId): IFlowNode?
+
+	/**
+	 * @return ID of the newly created entry
+	 */
+	suspend fun <S : NodeSettings> create(userID: UUID, descriptor: NodeDescriptor, settings: S): NodeId
+
+	/**
+	 * Connects two nodes together
+	 */
+	suspend fun connect(from: NodeId, to: NodeId): ServiceResult
+}
+
+object NodeServiceResults {
+	class NodeNotFoundError(val id: NodeId) : ServiceResult(HttpStatusCode.NotFound, "Node with ID $id not found")
+	class NodeConnected(from: NodeId, to: NodeId) : ServiceResult(HttpStatusCode.OK, "Connected $from to $to")
+}

integrations/api/src/main/kotlin/me/snoty/integration/common/IntegrationConfig.kt

@@ -1,14 +1,13 @@
 package me.snoty.integration.common.diff
 
 import kotlinx.coroutines.flow.Flow
-import me.snoty.integration.common.InstanceId
 import me.snoty.integration.common.diff.state.EntityState
-import java.util.UUID
+import me.snoty.integration.common.wiring.IFlowNode
 
 interface EntityStateService {
 	fun scheduleMetricsTask()
 
-	suspend fun updateStates(userID: UUID, instanceId: InstanceId, entities: List<IUpdatableEntity<out Any>>)
+	suspend fun updateStates(node: IFlowNode, entities: List<IUpdatableEntity<out Any>>)
 
-	fun getEntities(userID: UUID, instanceId: InstanceId, type: String): Flow<EntityState>
+	fun getStates(node: IFlowNode): Flow<EntityState>
 }

integrations/api/src/main/kotlin/me/snoty/integration/common/IntegrationDescriptor.kt

@@ -3,16 +3,14 @@ package me.snoty.integration.common.diff
 import com.mongodb.kotlin.client.coroutine.MongoCollection
 import kotlinx.datetime.Clock
 import kotlinx.datetime.Instant
-import me.snoty.integration.common.InstanceId
+import me.snoty.backend.integration.config.flow.NodeId
 import org.bson.codecs.pojo.annotations.BsonId
 import org.bson.types.ObjectId
-import java.util.*
 
 data class EntityDescriptor(
-	val instanceId: InstanceId,
+	val nodeId: NodeId,
 	val entityType: String,
 	val entityId: String,
-	val userId: UUID
 )
 
 data class UserEntityChanges(

integrations/api/src/main/kotlin/me/snoty/integration/common/IntegrationSettings.kt

@@ -1,10 +1,9 @@
 package me.snoty.integration.common.diff.state
 
 import com.mongodb.kotlin.client.coroutine.MongoCollection
-import me.snoty.integration.common.InstanceId
+import me.snoty.backend.integration.config.flow.NodeId
 import me.snoty.integration.common.diff.Fields
 import org.bson.codecs.pojo.annotations.BsonId
-import java.util.*
 
 data class EntityState(
 	val id: String,
@@ -13,10 +12,10 @@ data class EntityState(
 	val checksum: Long
 )
 
-data class UserEntityStates(
+data class NodeEntityStates(
 	@BsonId
-	val userId: UUID,
-	val entities: Map<InstanceId, Set<EntityState>>
+	val nodeId: NodeId,
+	val entities: Set<EntityState>
 )
 
-typealias EntityStateCollection = MongoCollection<UserEntityStates>
+typealias EntityStateCollection = MongoCollection<NodeEntityStates>

integrations/api/src/main/kotlin/me/snoty/integration/common/IntegrationContext.kt integrations/api/src/main/kotlin/me/snoty/integration/common/NodeContext.kt

@@ -1,22 +1,22 @@
 package me.snoty.integration.common.diff.state
 
 import com.mongodb.client.model.Filters
-import com.mongodb.client.model.Updates
-import me.snoty.integration.common.InstanceId
+import me.snoty.backend.integration.config.flow.NodeId
 import me.snoty.integration.common.diff.IUpdatableEntity
+import me.snoty.integration.common.wiring.IFlowNode
 import java.util.*
 
-suspend fun EntityStateCollection.updateStates(userId: UUID, instanceId: InstanceId, entities: List<IUpdatableEntity<out Any>>) {
+suspend fun EntityStateCollection.updateStates(node: IFlowNode, entities: List<IUpdatableEntity<out Any>>) {
 	val mapped = entities.mapTo(HashSet()) {
 		EntityState(it.id.toString(), it.type, it.fields, it.checksum)
 	}
-	updateStates(userId, instanceId, mapped)
+	updateStates(node._id, mapped)
 }
 
-suspend fun EntityStateCollection.updateStates(userId: UUID, instanceId: InstanceId, entities: Set<EntityState>) {
-	val update = Updates.set("${UserEntityStates::entities.name}.$instanceId", entities)
-	val result = updateOne(Filters.eq("_id", userId), update)
+suspend fun EntityStateCollection.updateStates(nodeId: NodeId, entities: Set<EntityState>) {
+	val states = NodeEntityStates(nodeId, entities)
+	val result = replaceOne(Filters.eq(nodeId), states)
 	if (result.matchedCount == 0L) {
-		insertOne(UserEntityStates(userId, mapOf(instanceId to entities)))
+		insertOne(states)
 	}
 }