1+ package streams.service.sink.strategy
2+
3+ import streams.events.EntityType
4+ import streams.extensions.quote
5+ import streams.serialization.JSONUtils
6+ import streams.service.StreamsSinkEntity
7+ import streams.service.sink.strategy.CUDIngestionStrategy.Companion.FROM_KEY
8+ import streams.service.sink.strategy.CUDIngestionStrategy.Companion.TO_KEY
9+ import streams.utils.IngestionUtils.getLabelsAsString
10+ import streams.utils.IngestionUtils.getNodeKeysAsString
11+ import streams.utils.StreamsUtils
12+
13+
14+ enum class CUDOperations { create, merge, update, delete }
15+
16+ abstract class CUD {
17+ abstract val op: CUDOperations
18+ abstract val type: EntityType
19+ abstract val properties: Map <String , Any ?>
20+ }
21+
22+ data class CUDNode (override val op : CUDOperations ,
23+ override val properties : Map <String , Any ?> = emptyMap(),
24+ val ids : Map <String , Any ?> = emptyMap(),
25+ val detach : Boolean = true ,
26+ val labels : List <String > = emptyList()): CUD() {
27+ override val type = EntityType .node
28+
29+ fun toMap (): Map <String , Any > {
30+ return when (op) {
31+ CUDOperations .delete -> mapOf (" ids"
32+ else -> mapOf (" ids" " properties"
33+ }
34+ }
35+ }
36+
37+ data class CUDNodeRel (val ids : Map <String , Any ?> = emptyMap(),
38+ val labels : List <String >)
39+ data class CUDRelationship (override val op : CUDOperations ,
40+ override val properties : Map <String , Any ?> = emptyMap(),
41+ val rel_type : String ,
42+ val from : CUDNodeRel ,
43+ val to : CUDNodeRel ): CUD() {
44+ override val type = EntityType .relationship
45+
46+ fun toMap (): Map <String , Any > {
47+ val from = mapOf (" ids"
48+ val to = mapOf (" ids"
49+ return when (op) {
50+ CUDOperations .delete -> mapOf (FROM_KEY to from,
51+ TO_KEY to to)
52+ else -> mapOf (FROM_KEY to from,
53+ TO_KEY to to,
54+ " properties"
55+ }
56+ }
57+ }
58+
59+
60+ class CUDIngestionStrategy : IngestionStrategy {
61+
62+ companion object {
63+ @JvmStatic val ID_KEY = " ids"
64+ @JvmStatic val PHYSICAL_ID_KEY = " _id"
65+ @JvmStatic val FROM_KEY = " from"
66+ @JvmStatic val TO_KEY = " to"
67+ }
68+
69+ data class NodeRelMetadata (val labels : List <String >, val ids : Set <String >)
70+
71+ private fun buildNodeLookupByIds (keyword : String = "MATCH ", ids : Set <String >, labels : List <String >, identifier : String = "n", field : String = ""): String {
72+ val fullField = if (field.isNotBlank()) " $field ." else field
73+ val quotedIdentifier = identifier.quote()
74+ return when (ids.contains(PHYSICAL_ID_KEY )) {
75+ true -> " MATCH ($quotedIdentifier ) WHERE id($quotedIdentifier ) = event.$fullField$ID_KEY ._id"
76+ else -> " $keyword ($quotedIdentifier${getLabelsAsString(labels)} {${getNodeKeysAsString(keys = ids, prefix = " $fullField$ID_KEY " })"
77+ }
78+ }
79+
80+ private fun buildNodeCreateStatement (labels : List <String >): String = """
81+ |${StreamsUtils .UNWIND }
82+ |CREATE (n${getLabelsAsString(labels)} )
83+ |SET n = event.properties
84+ """ .trimMargin()
85+
86+ private fun buildRelCreateStatement (from : NodeRelMetadata , to : NodeRelMetadata ,
87+ rel_type : String ): String = """
88+ |${StreamsUtils .UNWIND }
89+ |${buildNodeLookupByIds(ids = from.ids, labels = from.labels, identifier = FROM_KEY , field = FROM_KEY )}
90+ |${buildNodeLookupByIds(ids = to.ids, labels = to.labels, identifier = TO_KEY , field = TO_KEY )}
91+ |CREATE ($FROM_KEY )-[r:${rel_type.quote()} ]->($TO_KEY )
92+ |SET r = event.properties
93+ """ .trimMargin()
94+
95+ private fun buildNodeMergeStatement (labels : List <String >, ids : Set <String >): String = """
96+ |${StreamsUtils .UNWIND }
97+ |${buildNodeLookupByIds(keyword = " MERGE" = ids, labels = labels)}
98+ |SET n += event.properties
99+ """ .trimMargin()
100+
101+ private fun buildRelMergeStatement (from : NodeRelMetadata , to : NodeRelMetadata ,
102+ rel_type : String ): String = """
103+ |${StreamsUtils .UNWIND }
104+ |${buildNodeLookupByIds(ids = from.ids, labels = from.labels, identifier = FROM_KEY , field = FROM_KEY )}
105+ |${buildNodeLookupByIds(ids = to.ids, labels = to.labels, identifier = TO_KEY , field = TO_KEY )}
106+ |MERGE ($FROM_KEY )-[r:${rel_type.quote()} ]->($TO_KEY )
107+ |SET r += event.properties
108+ """ .trimMargin()
109+
110+ private fun buildNodeUpdateStatement (labels : List <String >, ids : Set <String >): String = """
111+ |${StreamsUtils .UNWIND }
112+ |${buildNodeLookupByIds(ids = ids, labels = labels)}
113+ |SET n += event.properties
114+ """ .trimMargin()
115+
116+ private fun buildRelUpdateStatement (from : NodeRelMetadata , to : NodeRelMetadata ,
117+ rel_type : String ): String = """
118+ |${StreamsUtils .UNWIND }
119+ |${buildNodeLookupByIds(ids = from.ids, labels = from.labels, identifier = FROM_KEY , field = FROM_KEY )}
120+ |${buildNodeLookupByIds(ids = to.ids, labels = to.labels, identifier = TO_KEY , field = TO_KEY )}
121+ |MATCH ($FROM_KEY )-[r:${rel_type.quote()} ]->($TO_KEY )
122+ |SET r += event.properties
123+ """ .trimMargin()
124+
125+ private fun buildDeleteStatement (labels : List <String >, ids : Set <String >, detach : Boolean ): String = """
126+ |${StreamsUtils .UNWIND }
127+ |${buildNodeLookupByIds(ids = ids, labels = labels)}
128+ |${if (detach) " DETACH " else " " DELETE n
129+ """ .trimMargin()
130+
131+ private fun buildRelDeleteStatement (from : NodeRelMetadata , to : NodeRelMetadata ,
132+ rel_type : String ): String = """
133+ |${StreamsUtils .UNWIND }
134+ |${buildNodeLookupByIds(ids = from.ids, labels = from.labels, identifier = FROM_KEY , field = FROM_KEY )}
135+ |${buildNodeLookupByIds(ids = to.ids, labels = to.labels, identifier = TO_KEY , field = TO_KEY )}
136+ |MATCH ($FROM_KEY )-[r:${rel_type.quote()} ]->($TO_KEY )
137+ |DELETE r
138+ """ .trimMargin()
139+
140+ private inline fun <reified T : CUD > toCUDEntity (it : Any ): T ? {
141+ return when (it) {
142+ is T -> it
143+ is Map <* , * > -> {
144+ val type = it[" type"
145+ val entityType = if (type == null ) null else EntityType .valueOf(type)
146+ when {
147+ entityType == null -> throw RuntimeException (" No `type` field found"
148+ entityType != null && EntityType .node == entityType && T ::class .java != CUDNode ::class .java -> null
149+ entityType != null && EntityType .relationship == entityType && T ::class .java != CUDRelationship ::class .java -> null
150+ else -> JSONUtils .convertValue<T >(it)
151+ }
152+ }
153+ else -> null
154+ }
155+ }
156+
157+ private fun getLabels (relNode : CUDNodeRel ) = if (relNode.ids.containsKey(PHYSICAL_ID_KEY )) emptyList() else relNode.labels
158+ private fun getLabels (node : CUDNode ) = if (node.ids.containsKey(PHYSICAL_ID_KEY )) emptyList() else node.labels
159+
160+ override fun mergeNodeEvents (events : Collection <StreamsSinkEntity >): List <QueryEvents > {
161+ val data = events
162+ .mapNotNull {
163+ it.value?.let {
164+ try {
165+ val data = toCUDEntity<CUDNode >(it)
166+ when (data?.op) {
167+ CUDOperations .delete, null -> null
168+ CUDOperations .merge -> if (data.ids.isNotEmpty() && data.properties.isNotEmpty()) data else null // TODO send to the DLQ the null
169+ else -> if (data.properties.isNotEmpty()) data else null // TODO send to the DLQ the null
170+ }
171+ } catch (e: Exception ) {
172+ null
173+ }
174+ }
175+ }
176+ .groupBy({ it.op }, { it })
177+
178+ val create = data[CUDOperations .create]
179+ .orEmpty()
180+ .groupBy { getLabels(it) }
181+ .map { QueryEvents (buildNodeCreateStatement(it.key), it.value.map { it.toMap() }) }
182+ val merge = data[CUDOperations .merge]
183+ .orEmpty()
184+ .groupBy { getLabels(it) to it.ids.keys }
185+ .map { QueryEvents (buildNodeMergeStatement(it.key.first, it.key.second), it.value.map { it.toMap() }) }
186+ val update = data[CUDOperations .update]
187+ .orEmpty()
188+ .groupBy { getLabels(it) to it.ids.keys }
189+ .map { QueryEvents (buildNodeUpdateStatement(it.key.first, it.key.second), it.value.map { it.toMap() }) }
190+ return (create + merge + update) // we'll group the data because of in case of `_id` key is present the generated queries are the same for update/merge
191+ .map { it.query to it.events }
192+ .groupBy({ it.first }, { it.second })
193+ .map { QueryEvents (it.key, it.value.flatten()) }
194+ }
195+
196+ override fun deleteNodeEvents (events : Collection <StreamsSinkEntity >): List <QueryEvents > {
197+ return events
198+ .mapNotNull {
199+ it.value?.let {
200+ try {
201+ val data = toCUDEntity<CUDNode >(it)
202+ when (data?.op) {
203+ CUDOperations .delete -> if (data.ids.isNotEmpty() && data.properties.isEmpty()) data else null // TODO send to the DLQ the null
204+ else -> null // TODO send to the DLQ the null
205+ }
206+ } catch (e: Exception ) {
207+ null
208+ }
209+ }
210+ }
211+ .groupBy { Triple (it.labels, it.ids.keys, it.detach) }
212+ .map {
213+ val (labels, keys, detach) = it.key
214+ QueryEvents (buildDeleteStatement(labels, keys, detach), it.value.map { it.toMap() })
215+ }
216+ }
217+
218+ override fun mergeRelationshipEvents (events : Collection <StreamsSinkEntity >): List <QueryEvents > {
219+ val data = events
220+ .mapNotNull {
221+ it.value?.let {
222+ try {
223+ val data = toCUDEntity<CUDRelationship >(it)
224+ when (data?.op) {
225+ CUDOperations .delete, null -> null // TODO send to the DLQ the null
226+ else -> if (data.from.ids.isNotEmpty() && data.to.ids.isNotEmpty() && data.properties.isNotEmpty()) data else null // TODO send to the DLQ the null
227+ }
228+ } catch (e: Exception ) {
229+ null
230+ }
231+ }
232+ }
233+ .groupBy({ it.op }, { it })
234+
235+ return data.flatMap { (op, list) ->
236+ list.groupBy { Triple (NodeRelMetadata (getLabels(it.from), it.from.ids.keys), NodeRelMetadata (getLabels(it.to), it.to.ids.keys), it.rel_type) }
237+ .map {
238+ val (from, to, rel_type) = it.key
239+ val query = when (op) {
240+ CUDOperations .create -> buildRelCreateStatement(from, to, rel_type)
241+ CUDOperations .merge -> buildRelMergeStatement(from, to, rel_type)
242+ else -> buildRelUpdateStatement(from, to, rel_type)
243+ }
244+ QueryEvents (query, it.value.map { it.toMap() })
245+ }
246+ }
247+ }
248+
249+ override fun deleteRelationshipEvents (events : Collection <StreamsSinkEntity >): List <QueryEvents > {
250+ return events
251+ .mapNotNull {
252+ it.value?.let {
253+ try {
254+ val data = toCUDEntity<CUDRelationship >(it)
255+ when (data?.op) {
256+ CUDOperations .delete -> if (data.from.ids.isNotEmpty() && data.to.ids.isNotEmpty() && data.properties.isEmpty()) data else null // TODO send to the DLQ the null
257+ else -> null // TODO send to the DLQ the null
258+ }
259+ } catch (e: Exception ) {
260+ null
261+ }
262+ }
263+ }
264+ .groupBy { Triple (NodeRelMetadata (getLabels(it.from), it.from.ids.keys), NodeRelMetadata (getLabels(it.to), it.to.ids.keys), it.rel_type) }
265+ .map {
266+ val (from, to, rel_type) = it.key
267+ QueryEvents (buildRelDeleteStatement(from, to, rel_type), it.value.map { it.toMap() })
268+ }
269+ }
270+
271+ }
0 commit comments