feat: working data download and indexing

src/lib/data-source.mjs

@@ -0,0 +1,80 @@
+import logger from './logger.mjs';
+import { BadRequest } from './es-errors.mjs';
+import zeroPad from './zero-pad.mjs';
+
+const doctype = 'doc';
+
+// helper for time-based indices
+function getIndexName(index) {
+  const d = new Date();
+  const [year, month, day] = [
+    d.getFullYear(),
+    zeroPad(d.getMonth() + 1, 2),
+    zeroPad(d.getDate(), 2),
+  ];
+  return `${index}-${year}.${month}.${day}`;
+}
+
+export async function createIndex(client, index) {
+  const realIndex = getIndexName(index);
+  return client.indices
+    .create({
+      index: realIndex,
+      body: {
+        settings: {},
+        mappings: {
+          [doctype]: {
+            properties: {
+              transponder: { type: 'keyword' },
+              callsign: { type: 'keyword' },
+              origin_country: { type: 'keyword' },
+              time_position: { type: 'date' },
+              last_contact: { type: 'date' },
+              location: { type: 'geo_point' },
+              baro_altitude: { type: 'float' },
+              geo_altitude: { type: 'float' },
+              on_ground: { type: 'boolean' },
+              velocity: { type: 'float' },
+              vertical_rate: { type: 'float' },
+              squawk: { type: 'keyword' },
+              spi: { type: 'boolean' },
+              position_source: { type: 'keyword' },
+            },
+          },
+        },
+      },
+    })
+    .then(res => {
+      logger.debug(`Index created: ${res.index}`);
+      return res.index;
+    })
+    .catch(err => {
+      // check for existing index
+      if (err instanceof BadRequest) {
+        logger.debug(`Index exists: ${realIndex}`);
+        return client.indices.get({ index: realIndex }).then(() => realIndex);
+      }
+
+      throw err;
+    });
+}
+
+export async function createDocument(client, index, body) {
+  client.index({
+    index,
+    type: doctype,
+    body,
+  });
+}
+
+export async function bulkInsert(client, index, docs) {
+  const body = docs.reduce((collection, doc) => {
+    collection.push({ index: { _index: index, _type: doctype } });
+    collection.push(doc);
+    return collection;
+  }, []);
+
+  client.bulk({
+    body,
+  });
+}