import locations to analytics database on reindex

Author: rikukissa

infrastructure/postgres/setup-analytics.sh

@@ -50,31 +50,35 @@ PGPASSWORD="$POSTGRES_PASSWORD" psql -v ON_ERROR_STOP=1 -h "$POSTGRES_HOST" -p "
 
 CREATE SCHEMA IF NOT EXISTS analytics;
 
-CREATE OR REPLACE VIEW analytics.locations
-WITH (security_barrier)
-AS
-SELECT * FROM app.locations;
+CREATE TABLE IF NOT EXISTS analytics.locations (
+  id uuid PRIMARY KEY,
+  name text NOT NULL,
+  parent_id uuid REFERENCES analytics.locations(id),
+  location_type TEXT NOT NULL
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS analytics_locations_pkey ON analytics.locations(id uuid_ops);
 
 CREATE TABLE IF NOT EXISTS analytics.event_actions (
   event_type text NOT NULL,
-  action_type app.action_type NOT NULL,
+  action_type TEXT NOT NULL,
   annotation jsonb,
   assigned_to text,
   created_at timestamp with time zone NOT NULL DEFAULT now(),
-  created_at_location uuid REFERENCES app.locations(id),
+  created_at_location uuid,
   created_by text NOT NULL,
   created_by_role text NOT NULL,
   created_by_signature text,
-  created_by_user_type app.user_type NOT NULL,
+  created_by_user_type TEXT NOT NULL,
   declared_at timestamp with time zone,
   registered_at timestamp with time zone,
   declaration jsonb NOT NULL DEFAULT '{}'::jsonb,
-  event_id uuid NOT NULL REFERENCES app.events(id),
+  event_id uuid NOT NULL,
   id uuid DEFAULT gen_random_uuid() PRIMARY KEY,
-  original_action_id uuid REFERENCES app.event_actions(id),
+  original_action_id uuid,
   registration_number text UNIQUE,
   request_id text,
-  status app.action_status NOT NULL,
+  status TEXT NOT NULL,
   transaction_id text NOT NULL,
   content jsonb,
   UNIQUE (id, event_id)

src/analytics/analytics.ts

@@ -24,7 +24,8 @@ import {
   EventDocument,
   EventState,
   getActionAnnotationFields,
-  getCurrentEventState
+  getCurrentEventState,
+  Location
 } from '@opencrvs/toolkit/events'
 import { differenceInDays } from 'date-fns'
 import { ExpressionBuilder, Kysely } from 'kysely'
@@ -232,6 +233,43 @@ export async function importEvent(event: EventDocument, trx: Kysely<any>) {
   logger.info(`Event with id "${event.id}" logged into analytics`)
 }
 
+export async function importLocations(locations: Location[]) {
+  const client = getClient()
+  await client.transaction().execute(async (trx) => {
+    for (const [index, batch] of chunk(
+      locations,
+      INSERT_MAX_CHUNK_SIZE
+    ).entries()) {
+      logger.info(
+        `Importing ${Math.min((index + 1) * INSERT_MAX_CHUNK_SIZE, locations.length)}/${locations.length} locations`
+      )
+
+      await trx
+        .insertInto('analytics.locations')
+        .values(
+          batch.map((l) => ({
+            id: l.id,
+            name: l.name,
+            parentId: l.parentId,
+            locationType: l.locationType
+          }))
+        )
+        .onConflict((oc) =>
+          oc
+            .column('id')
+            .doUpdateSet(
+              (eb: ExpressionBuilder<any, 'analytics.locations'>) => ({
+                name: eb.ref('excluded.name'),
+                parentId: eb.ref('excluded.parentId'),
+                locationType: eb.ref('excluded.locationType')
+              })
+            )
+        )
+        .execute()
+    }
+  })
+}
+
 export async function importEvents(events: EventDocument[], trx: Kysely<any>) {
   for (const event of events) {
     await importEvent(event, trx)

src/index.ts

@@ -30,7 +30,8 @@ import {
   COUNTRY_CONFIG_PORT,
   CHECK_INVALID_TOKEN,
   AUTH_URL,
-  DEFAULT_TIMEOUT
+  DEFAULT_TIMEOUT,
+  GATEWAY_URL
 } from '@countryconfig/constants'
 import {
   contentHandler,
@@ -70,11 +71,13 @@ import getUserNotificationRoutes from './config/routes/userNotificationRoutes'
 import {
   importEvent,
   importEvents,
+  importLocations,
   syncLocationLevels,
   syncLocationStatistics
 } from './analytics/analytics'
 import { getClient } from './analytics/postgres'
 import { env } from './environment'
+import { createClient } from '@opencrvs/toolkit/api'
 
 export interface ITokenPayload {
   sub: string
@@ -605,6 +608,12 @@ export async function createServer() {
           if (queue.length > 0) {
             await importEvents(queue, trx)
           }
+
+          // Import locations
+          const url = new URL('events', GATEWAY_URL).toString()
+          const client = createClient(url, req.headers.authorization)
+          const locations = await client.locations.list.query()
+          await importLocations(locations)
         })
 
         logger.info('Reindexed all events into analytics.')
@@ -687,6 +696,7 @@ export async function createServer() {
     const parsedPath = /^\/trigger\/events\/[^/]+\/actions\/([^/]+)$/.exec(
       request.route.path
     )
+
     const actionType = parsedPath?.[1] as ActionType | null
     const wasRequestForActionConfirmation =
       actionType && request.method === 'post'
@@ -695,9 +705,15 @@ export async function createServer() {
     if (wasRequestForActionConfirmation && wasActionAcceptedImmediately) {
       const event = request.payload as EventDocument
       const client = getClient()
-      await client.transaction().execute(async (trx) => {
-        await importEvent(event, trx)
-      })
+      try {
+        await client.transaction().execute(async (trx) => {
+          await importEvent(event, trx)
+        })
+      } catch (error) {
+        // eslint-disable-next-line no-console
+        console.error(error)
+        throw error
+      }
     }
     return h.continue
   })