Search on es

discovery-provider/.test.env

@@ -1,4 +1,7 @@
 # TODO: dummy ganache keys for local setup; should wire with dynamically generated keys
 audius_delegate_owner_wallet=0x1D9c77BcfBfa66D37390BF2335f0140979a6122B
 audius_delegate_private_key=0x3873ed01bfb13621f9301487cc61326580614a5b99f3c33cf39c6f9da3a19cad
-audius_solana_rewards_manager_account=8MzNUaBHskteN7poTrZG5wgSNSbXQwieMDB4wk9fgB7f
+audius_solana_rewards_manager_account=8MzNUaBHskteN7poTrZG5wgSNSbXQwieMDB4wk9fgB7f
+
+audius_elasticsearch_url=http://localhost:9200
+audius_db_url=postgresql+psycopg2://postgres:postgres@localhost:5432/test_audius_discovery

discovery-provider/compose/docker-compose.backend.yml

@@ -13,6 +13,9 @@ services:
       - audius_redis_url=redis://${COMPOSE_PROJECT_NAME}_redis-server_1:6379/00
       - audius_db_url=postgresql+psycopg2://postgres:postgres@${COMPOSE_PROJECT_NAME}_discovery-provider-db_1:5432/audius_discovery
       - audius_db_url_read_replica=postgresql+psycopg2://postgres:postgres@${COMPOSE_PROJECT_NAME}_discovery-provider-db_1:5432/audius_discovery
+      - audius_elasticsearch_url=http://elasticsearch:9200
+      - audius_elasticsearch_run_indexer=true
+      - audius_elasticsearch_search_enabled=true
       - audius_delegate_owner_wallet=${audius_delegate_owner_wallet}
       - audius_delegate_private_key=${audius_delegate_private_key}
       - audius_ipfs_host=${COMPOSE_PROJECT_NAME}-ipfs-node

discovery-provider/es-indexer/README.md

@@ -21,6 +21,22 @@ If you are adding a new denormalization (attaching data from a related model), t
 - For "catchup" mode this is the `checkpointSql` function. See UserIndexer or TrackIndexer for an example
 - For listen / notify mode, this is the handler code in `listen.ts`
 
+When working on mapping changes, I might put code like this at top of `main.ts main()` function:
+
+```ts
+await new Promise((r) => setTimeout(r, 100)) // don't ask... will fix haha
+await indexer.playlists.createIndex({ drop: true })
+await indexer.playlists.catchup()
+process.exit(0)
+```
+
+and then:
+
+```
+source .env
+npm run dev
+```
+
 ## How it works
 
 Program attempts to avoid any gaps by doing a "catchup" on boot... when complete it swithces to processing "batches" which are events collected from postgres LISTEN / NOTIFY.
@@ -43,8 +59,7 @@ Check "elasticsearch" health info in `/health_check?verbose=true` endpoint.
 (instructions for sandbox3... subject to change):
 
 Use Kibana:
-Uncomment the kibana container and restart discovery-provider. 
-
+Uncomment the kibana container and restart discovery-provider.
 
 List indices:
 

discovery-provider/es-indexer/package.json

@@ -6,7 +6,7 @@
     "dev": "ts-node src/main.ts",
     "nuke": "ts-node nuke.ts",
     "start": "tsc && pm2-runtime build/src/main.js --restart-delay=3000",
-    "test": "echo \"Error: no test specified\" && exit 1"
+    "test": "tsc --noEmit"
   },
   "keywords": [],
   "author": "",

discovery-provider/es-indexer/src/conn.ts

@@ -43,6 +43,14 @@ export async function waitForHealthyCluster() {
   )
 }
 
+export async function ensureSaneCluterSettings() {
+  return dialEs().cluster.putSettings({
+    persistent: {
+      'action.auto_create_index': false,
+    },
+  })
+}
+
 /**
  * Gets the max(blocknumber) from elasticsearch indexes
  * Used for incremental indexing to understand "where we were" so we can load new data from postgres

discovery-provider/es-indexer/src/indexNames.ts

@@ -1,7 +1,7 @@
 export const indexNames = {
-  playlists: 'playlists2',
+  playlists: 'playlists6',
   reposts: 'reposts2',
   saves: 'saves2',
-  tracks: 'tracks2',
-  users: 'users2',
+  tracks: 'tracks6',
+  users: 'users6',
 }

discovery-provider/es-indexer/src/indexers/BaseIndexer.ts

@@ -44,6 +44,12 @@ export abstract class BaseIndexer<RowType> {
     }
   }
 
+  async refreshIndex() {
+    const { es, logger, indexName } = this
+    logger.info('refreshing index: ' + indexName)
+    await es.indices.refresh({ index: indexName })
+  }
+
   async cutoverAlias() {
     const { es, logger, indexName, tableName } = this
 
@@ -173,7 +179,7 @@ export abstract class BaseIndexer<RowType> {
     ])
   }
 
-  async withBatch(rows: Array<RowType>) {}
+  async withBatch(rows: Array<RowType>) { }
 
-  withRow(row: RowType) {}
+  withRow(row: RowType) { }
 }

discovery-provider/es-indexer/src/indexers/PlaylistIndexer.ts

@@ -1,10 +1,15 @@
 import { IndicesCreateRequest } from '@elastic/elasticsearch/lib/api/types'
-import { keyBy } from 'lodash'
+import { keyBy, merge } from 'lodash'
 import { dialPg } from '../conn'
 import { indexNames } from '../indexNames'
 import { BlocknumberCheckpoint } from '../types/blocknumber_checkpoint'
 import { PlaylistDoc } from '../types/docs'
 import { BaseIndexer } from './BaseIndexer'
+import {
+  sharedIndexSettings,
+  standardSuggest,
+  standardText,
+} from './sharedIndexSettings'
 
 export class PlaylistIndexer extends BaseIndexer<PlaylistDoc> {
   tableName = 'playlists'
@@ -13,13 +18,13 @@ export class PlaylistIndexer extends BaseIndexer<PlaylistDoc> {
 
   mapping: IndicesCreateRequest = {
     index: indexNames.playlists,
-    settings: {
+    settings: merge(sharedIndexSettings, {
       index: {
         number_of_shards: 1,
         number_of_replicas: 0,
         refresh_interval: '5s',
       },
-    },
+    }),
     mappings: {
       dynamic: false,
       properties: {
@@ -30,9 +35,37 @@ export class PlaylistIndexer extends BaseIndexer<PlaylistDoc> {
         is_album: { type: 'boolean' },
         is_private: { type: 'boolean' },
         is_delete: { type: 'boolean' },
-        playlist_name: { type: 'text' },
+        suggest: standardSuggest,
+        playlist_name: {
+          type: 'keyword',
+          fields: {
+            searchable: standardText,
+          },
+        },
         'playlist_contents.track_ids.track': { type: 'keyword' },
 
+        user: {
+          properties: {
+            handle: {
+              type: 'keyword',
+              fields: {
+                searchable: standardText,
+              },
+            },
+            name: {
+              type: 'keyword',
+              fields: {
+                searchable: standardText,
+              },
+            },
+            location: { type: 'keyword' },
+            follower_count: { type: 'integer' },
+            is_verified: { type: 'boolean' },
+            created_at: { type: 'date' },
+            updated_at: { type: 'date' },
+          },
+        },
+
         // saves
         saved_by: { type: 'keyword' },
         save_count: { type: 'integer' },
@@ -58,7 +91,17 @@ export class PlaylistIndexer extends BaseIndexer<PlaylistDoc> {
     return `
       -- etl playlists
       select 
-        *,
+        playlists.*,
+
+        json_build_object(
+          'handle', users.handle,
+          'name', users.name,
+          'location', users.location,
+          'follower_count', follower_count,
+          'is_verified', users.is_verified,
+          'created_at', users.created_at,
+          'updated_at', users.updated_at
+        ) as user,
 
         array(
           select user_id 
@@ -83,7 +126,11 @@ export class PlaylistIndexer extends BaseIndexer<PlaylistDoc> {
         ) as saved_by
 
       from playlists 
-      where is_current = true
+      join users on playlist_owner_id = user_id
+      left join aggregate_user on users.user_id = aggregate_user.user_id
+      where 
+        playlists.is_current
+        AND users.is_current
     `
   }
 
@@ -132,6 +179,9 @@ export class PlaylistIndexer extends BaseIndexer<PlaylistDoc> {
   }
 
   withRow(row: PlaylistDoc) {
+    row.suggest = [row.playlist_name, row.user.handle, row.user.name]
+      .filter((x) => x)
+      .join(' ')
     row.repost_count = row.reposted_by.length
     row.save_count = row.saved_by.length
   }