1 package de.dlr.shepard.common.neo4j.migrations;
2
3 import ac.simons.neo4j.migrations.core.JavaBasedMigration;
4 import ac.simons.neo4j.migrations.core.MigrationContext;
5 import com.fasterxml.jackson.core.JsonProcessingException;
6 import com.fasterxml.jackson.databind.JsonNode;
7 import com.fasterxml.jackson.databind.ObjectMapper;
8 import com.fasterxml.jackson.databind.util.StdDateFormat;
9 import io.quarkus.logging.Log;
10 import java.text.ParseException;
11 import java.util.Date;
12 import java.util.HashMap;
13 import java.util.Map;
14 import java.util.Optional;
15 import org.neo4j.driver.Session;
16
17 public class V2__Extract_json implements JavaBasedMigration {
18
19 record ShepardFile(String oid, String filename, long createdAt, String md5) {}
20
21 record StructuredData(String oid, String name, long createdAt) {}
22
23 record Timeseries(String measurement, String device, String location, String symbolicName, String field) {}
24
25 private static final String FILES_JSON = "filesJson";
26 private static final String STRUCTURED_DATAS_JSON = "structuredDatasJson";
27 private static final String TIMESERIES_JSON = "timeseriesJson";
28 private final ObjectMapper mapper = new ObjectMapper();
29
30 @SuppressWarnings("PMD.AvoidCatchingGenericException")
31 @Override
32 public void apply(MigrationContext context) {
33 try (Session session = context.getSession()) {
34 Log.info("Running V2 migration (1/5)");
35 migrateFileContainer(session);
36 Log.info("Running V2 migration (2/5)");
37 migrateFileReferences(session);
38 Log.info("Running V2 migration (3/5)");
39 migrateStructuredDataContainer(session);
40 Log.info("Running V2 migration (4/5)");
41 migrateStructuredDataReferences(session);
42 Log.info("Running V2 migration (5/5)");
43 migrateTimeseriesReferences(session);
44 } catch (Exception e) {
45 Log.error("Error while running migration: ", e);
46 }
47 }
48
49 @SuppressWarnings("PMD.AvoidDuplicateLiterals")
50 private void migrateFileContainer(Session session) {
51 var cResults = session.executeRead(tx ->
52 tx.run("MATCH (c:FileContainer) WHERE c.filesJson IS NOT NULL RETURN c").list()
53 );
54 for (int i = 0; i < cResults.size(); i++) {
55 logPercent(i, cResults.size());
56 var c = cResults.get(i).get("c").asNode();
57 var cId = c.elementId();
58
59 if (!c.containsKey(FILES_JSON)) continue;
60
61 try (var tx = session.beginTransaction()) {
62 for (var fileObj : c.get(FILES_JSON).asList()) {
63 if (fileObj instanceof String fileStr) {
64 var fileNode = parseJson(fileStr);
65 if (fileNode.isEmpty()) {
66 Log.errorf("NodeID %s: File cannot be parsed and will be skipped: %s", cId, fileStr);
67 continue;
68 }
69 var file = parseShepardFile(fileNode.get());
70 Map<String, Object> params = new HashMap<>();
71 params.put(
72 "props",
73 Map.of("oid", file.oid, "createdAt", file.createdAt, "filename", file.filename, "md5", file.md5)
74 );
75 var query =
76 """
77 MATCH (c:FileContainer) WHERE ID(c) = %s
78 CREATE (c)-[:file_in_container]->(sf:ShepardFile $props)
79 """;
80 tx.run(query.formatted(cId), params);
81 }
82 }
83 tx.run("MATCH (c:FileContainer) WHERE ID(c) = " + cId + " REMOVE c.filesJson");
84 tx.commit();
85 }
86 }
87 }
88
89 private void migrateFileReferences(Session session) {
90 var rResults = session.executeRead(tx ->
91 tx.run("MATCH (r:FileReference) WHERE r.filesJson IS NOT NULL RETURN r").list()
92 );
93 for (int i = 0; i < rResults.size(); i++) {
94 logPercent(i, rResults.size());
95 var r = rResults.get(i).get("r").asNode();
96 var rId = r.elementId();
97
98 if (!r.containsKey(FILES_JSON)) continue;
99
100 try (var tx = session.beginTransaction()) {
101 for (var fileObj : r.get(FILES_JSON).asList()) {
102 if (fileObj instanceof String fileStr) {
103 var fileNode = parseJson(fileStr);
104 if (fileNode.isEmpty()) {
105 Log.errorf("NodeID %s: File cannot be parsed and will be skipped: %s", rId, fileStr);
106 continue;
107 }
108 var file = parseShepardFile(fileNode.get());
109 Map<String, Object> params = new HashMap<>();
110 params.put("oid", file.oid);
111 params.put("props", Map.of("createdAt", file.createdAt, "filename", file.filename, "md5", file.md5));
112 var query =
113 """
114 MATCH (r:FileReference)-[:is_in_container]->(c:FileContainer) WHERE ID(r) = %s
115 MERGE (c)-[:file_in_container]->(sf:ShepardFile { oid: $oid })
116 SET sf += $props
117 CREATE (r)-[hp:has_payload]->(sf)
118 """;
119 tx.run(query.formatted(rId), params);
120 }
121 }
122 tx.run("MATCH (r:FileReference) WHERE ID(r) = " + rId + " REMOVE r.filesJson");
123 tx.commit();
124 }
125 }
126 }
127
128 private void migrateStructuredDataContainer(Session session) {
129 var cResults = session.executeRead(tx ->
130 tx.run("MATCH (c:StructuredDataContainer) WHERE c.structuredDatasJson IS NOT NULL RETURN c").list()
131 );
132 for (int i = 0; i < cResults.size(); i++) {
133 logPercent(i, cResults.size());
134 var c = cResults.get(i).get("c").asNode();
135 var cId = c.elementId();
136
137 if (!c.containsKey(STRUCTURED_DATAS_JSON)) continue;
138
139 try (var tx = session.beginTransaction();) {
140 for (var structuredDataObj : c.get(STRUCTURED_DATAS_JSON).asList()) {
141 if (structuredDataObj instanceof String structuredDataStr) {
142 var structuredDataNode = parseJson(structuredDataStr);
143 if (structuredDataNode.isEmpty()) {
144 Log.errorf("NodeID %s: StructuredData cannot be parsed and will be skipped: %s", cId, structuredDataStr);
145 continue;
146 }
147 var sd = parseStructuredData(structuredDataNode.get());
148 Map<String, Object> params = new HashMap<>();
149 params.put("props", Map.of("oid", sd.oid, "createdAt", sd.createdAt, "name", sd.name));
150 var query =
151 """
152 MATCH (c:StructuredDataContainer) WHERE ID(c) = %s
153 CREATE (c)-[:structureddata_in_container]->(sd:StructuredData $props)
154 """;
155 tx.run(query.formatted(cId), params);
156 }
157 }
158 tx.run("MATCH (c:StructuredDataContainer) WHERE ID(c) = " + cId + " REMOVE c.structuredDatasJson");
159 tx.commit();
160 }
161 }
162 }
163
164 private void migrateStructuredDataReferences(Session session) {
165 var rResults = session.executeRead(tx ->
166 tx.run("MATCH (r:StructuredDataReference) WHERE r.structuredDatasJson IS NOT NULL RETURN r").list()
167 );
168 for (int i = 0; i < rResults.size(); i++) {
169 logPercent(i, rResults.size());
170 var r = rResults.get(i).get("r").asNode();
171 var rId = r.elementId();
172
173 if (!r.containsKey(STRUCTURED_DATAS_JSON)) continue;
174
175 try (var tx = session.beginTransaction();) {
176 for (var structuredDataObj : r.get(STRUCTURED_DATAS_JSON).asList()) {
177 if (structuredDataObj instanceof String structuredDataStr) {
178 var structuredDataNode = parseJson(structuredDataStr);
179 if (structuredDataNode.isEmpty()) {
180 Log.errorf("NodeID %s: StructuredData cannot be parsed and will be skipped: %s", rId, structuredDataStr);
181 continue;
182 }
183 var sd = parseStructuredData(structuredDataNode.get());
184 Map<String, Object> params = new HashMap<>();
185 params.put("oid", sd.oid);
186 params.put("props", Map.of("createdAt", sd.createdAt, "name", sd.name));
187 var query =
188 """
189 MATCH (r:StructuredDataReference)-[:is_in_container]->(c:StructuredDataContainer) WHERE ID(r) = %s
190 MERGE (c)-[:structureddata_in_container]->(sd:StructuredData { oid: $oid })
191 SET sd += $props
192 CREATE (r)-[hp:has_payload]->(sd)
193 """;
194 tx.run(query.formatted(rId), params);
195 }
196 }
197 tx.run("MATCH (r:StructuredDataReference) WHERE ID(r) = " + rId + " REMOVE r.structuredDatasJson");
198 tx.commit();
199 }
200 }
201 }
202
203 private void migrateTimeseriesReferences(Session session) {
204 var rResults = session.executeRead(tx ->
205 tx.run("MATCH (r:TimeseriesReference) WHERE r.timeseriesJson IS NOT NULL RETURN r").list()
206 );
207 for (int i = 0; i < rResults.size(); i++) {
208 logPercent(i, rResults.size());
209 var r = rResults.get(i).get("r").asNode();
210 var rId = r.elementId();
211
212 if (!r.containsKey(TIMESERIES_JSON)) continue;
213
214 try (var tx = session.beginTransaction();) {
215 for (var timeseriesObj : r.get(TIMESERIES_JSON).asList()) {
216 if (timeseriesObj instanceof String timeseriesStr) {
217 var timeseriesNode = parseJson(timeseriesStr);
218 if (timeseriesNode.isEmpty()) {
219 Log.errorf("NodeID %s: Timeseries cannot be parsed and will be skipped: %s", rId, timeseriesStr);
220 continue;
221 }
222 var ts = parseTimeseries(timeseriesNode.get());
223 Map<String, Object> params = Map.of(
224 "measurement",
225 ts.measurement,
226 "device",
227 ts.device,
228 "location",
229 ts.location,
230 "symbolicName",
231 ts.symbolicName,
232 "field",
233 ts.field
234 );
235 var query =
236 """
237 MATCH (r:TimeseriesReference) WHERE ID(r) = %s
238 MERGE (ts:Timeseries { measurement: $measurement, device: $device, location: $location, symbolicName: $symbolicName, field: $field })
239 CREATE (r)-[hp:has_payload]->(ts)
240 """;
241
242 tx.run(query.formatted(rId), params);
243 }
244 }
245 tx.run("MATCH (r:TimeseriesReference) WHERE ID(r) = " + rId + " REMOVE r.timeseriesJson");
246 tx.commit();
247 }
248 }
249 }
250
251 private Optional<JsonNode> parseJson(String str) {
252 JsonNode node;
253 try {
254 node = mapper.readTree(str);
255 } catch (JsonProcessingException e) {
256
257 Log.error(e.toString());
258 return Optional.empty();
259 }
260 return Optional.of(node);
261 }
262
263 private long parseDate(String date) {
264 if (date.isEmpty()) return 0L;
265 Date parsed;
266 try {
267 parsed = new StdDateFormat().parse(date);
268 } catch (ParseException e) {
269
270 Log.warnf("%s, using 0 instead", e.getMessage());
271 return 0L;
272 }
273 return parsed.getTime();
274 }
275
276 private ShepardFile parseShepardFile(JsonNode node) {
277 var oid = Optional.ofNullable(node.get("oid")).map(JsonNode::asText).orElse("");
278 var createdAt = Optional.ofNullable(node.get("createdAt")).map(JsonNode::asText).orElse("");
279 var filename = Optional.ofNullable(node.get("filename")).map(JsonNode::asText).orElse("");
280 var md5 = Optional.ofNullable(node.get("md5")).map(JsonNode::asText).orElse("");
281 return new ShepardFile(oid, filename, parseDate(createdAt), md5);
282 }
283
284 private StructuredData parseStructuredData(JsonNode node) {
285 var oid = Optional.ofNullable(node.get("oid")).map(JsonNode::asText).orElse("");
286 var createdAt = Optional.ofNullable(node.get("createdAt")).map(JsonNode::asText).orElse("");
287 var name = Optional.ofNullable(node.get("name")).map(JsonNode::asText).orElse("");
288 return new StructuredData(oid, name, parseDate(createdAt));
289 }
290
291 private Timeseries parseTimeseries(JsonNode node) {
292 var measurement = Optional.ofNullable(node.get("measurement")).map(JsonNode::asText).orElse("");
293 var device = Optional.ofNullable(node.get("device")).map(JsonNode::asText).orElse("");
294 var location = Optional.ofNullable(node.get("location")).map(JsonNode::asText).orElse("");
295 var symbolicName = Optional.ofNullable(node.get("symbolicName")).map(JsonNode::asText).orElse("");
296 var field = Optional.ofNullable(node.get("field")).map(JsonNode::asText).orElse("");
297 return new Timeseries(measurement, device, location, symbolicName, field);
298 }
299
300 private void logPercent(int i, int size) {
301 int curPercent = (int) ((100f / size) * i);
302 int prePercent = (int) ((100f / size) * (i - 1));
303 if (prePercent < curPercent) {
304 Log.infof("... %d %", curPercent);
305 }
306 }
307 }