Improvement/stricter serp api enrichment filter (#2670)

Author: epipav

services/apps/premium/members_enrichment_worker/src/activities.ts

@@ -1,5 +1,6 @@
 import {
   findMemberEnrichmentCache,
+  findMemberIdentityWithTheMostActivityInPlatform,
   getEnrichmentData,
   insertMemberEnrichmentCache,
   isCacheObsolete,
@@ -50,4 +51,5 @@ export {
   touchMemberEnrichmentCacheUpdatedAt,
   updateMemberEnrichmentCache,
   isEnrichableBySource,
+  findMemberIdentityWithTheMostActivityInPlatform,
 }

services/apps/premium/members_enrichment_worker/src/activities/enrichment.ts

@@ -1,10 +1,16 @@
+import { findMemberIdentityWithTheMostActivityInPlatform as findMemberIdentityWithTheMostActivityInPlatformQuestDb } from '@crowd/data-access-layer/src/activities'
 import {
   findMemberEnrichmentCacheDb,
   insertMemberEnrichmentCacheDb,
   touchMemberEnrichmentCacheUpdatedAtDb,
   updateMemberEnrichmentCacheDb,
 } from '@crowd/data-access-layer/src/old/apps/premium/members_enrichment_worker'
-import { IMemberEnrichmentCache, MemberEnrichmentSource } from '@crowd/types'
+import { RedisCache } from '@crowd/redis'
+import {
+  IEnrichableMemberIdentityActivityAggregate,
+  IMemberEnrichmentCache,
+  MemberEnrichmentSource,
+} from '@crowd/types'
 
 import { EnrichmentSourceServiceFactory } from '../factory'
 import { svc } from '../main'
@@ -27,7 +33,7 @@ export async function getEnrichmentData(
   input: IEnrichmentSourceInput,
 ): Promise<IMemberEnrichmentData | null> {
   const service = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, svc.log)
-  if (service.isEnrichableBySource(input)) {
+  if (service.isEnrichableBySource(input) && (await hasRemainingCredits(source))) {
     return service.getData(input)
   }
   return null
@@ -52,6 +58,41 @@ export async function isCacheObsolete(
   )
 }
 
+export async function setHasRemainingCredits(
+  source: MemberEnrichmentSource,
+  hasCredits: boolean,
+): Promise<void> {
+  const redisCache = new RedisCache(`enrichment-${source}`, svc.redis, svc.log)
+  if (hasCredits) {
+    await redisCache.set('hasRemainingCredits', 'true', 60)
+  } else {
+    await redisCache.set('hasRemainingCredits', 'false', 60)
+  }
+}
+
+export async function getHasRemainingCredits(source: MemberEnrichmentSource): Promise<boolean> {
+  const redisCache = new RedisCache(`enrichment-${source}`, svc.redis, svc.log)
+  return (await redisCache.get('hasRemainingCredits')) === 'true'
+}
+
+export async function hasRemainingCreditsExists(source: MemberEnrichmentSource): Promise<boolean> {
+  const redisCache = new RedisCache(`enrichment-${source}`, svc.redis, svc.log)
+  return await redisCache.exists('hasRemainingCredits')
+}
+
+export async function hasRemainingCredits(source: MemberEnrichmentSource): Promise<boolean> {
+  const service = EnrichmentSourceServiceFactory.getEnrichmentSourceService(source, svc.log)
+
+  if (await hasRemainingCreditsExists(source)) {
+    return getHasRemainingCredits(source)
+  }
+
+  const hasCredits = await service.hasRemainingCredits()
+
+  await setHasRemainingCredits(source, hasCredits)
+  return hasCredits
+}
+
 export async function findMemberEnrichmentCache(
   source: MemberEnrichmentSource,
   memberId: string,
@@ -81,3 +122,10 @@ export async function touchMemberEnrichmentCacheUpdatedAt(
 ): Promise<void> {
   await touchMemberEnrichmentCacheUpdatedAtDb(svc.postgres.writer.connection(), memberId, source)
 }
+
+export async function findMemberIdentityWithTheMostActivityInPlatform(
+  memberId: string,
+  platform: string,
+): Promise<IEnrichableMemberIdentityActivityAggregate> {
+  return findMemberIdentityWithTheMostActivityInPlatformQuestDb(svc.questdbSQL, memberId, platform)
+}

services/apps/premium/members_enrichment_worker/src/activities/getMembers.ts

@@ -11,7 +11,7 @@ import { svc } from '../main'
 export async function getEnrichableMembers(
   limit: number,
   sources: MemberEnrichmentSource[],
-  afterId: string,
+  afterCursor: { activityCount: number; memberId: string } | null,
 ): Promise<IEnrichableMember[]> {
   let rows: IEnrichableMember[] = []
   const sourceInputs: IMemberEnrichmentSourceQueryInput[] = sources.map((s) => {
@@ -23,7 +23,7 @@ export async function getEnrichableMembers(
     }
   })
   const db = svc.postgres.reader
-  rows = await fetchMembersForEnrichment(db, limit, sourceInputs, afterId)
+  rows = await fetchMembersForEnrichment(db, limit, sourceInputs, afterCursor)
 
   return rows
 }

services/apps/premium/members_enrichment_worker/src/schedules/getMembersToEnrich.ts

@@ -27,7 +27,7 @@ export const scheduleMembersEnrichment = async () => {
         },
         args: [
           {
-            afterId: null,
+            afterCursor: null,
           },
         ],
       },

services/apps/premium/members_enrichment_worker/src/sources/clearbit/service.ts

@@ -27,7 +27,9 @@ import {
 export default class EnrichmentServiceClearbit extends LoggerBase implements IEnrichmentService {
   public source: MemberEnrichmentSource = MemberEnrichmentSource.CLEARBIT
   public platform = `enrichment-${this.source}`
-  public enrichableBySql = `mi.type = 'email' and mi.verified`
+  public enrichMembersWithActivityMoreThan = 10
+
+  public enrichableBySql = `"activitySummary".total_count > ${this.enrichMembersWithActivityMoreThan} AND mi.type = 'email' and mi.verified`
 
   // bust cache after 120 days
   public cacheObsoleteAfterSeconds = 60 * 60 * 24 * 120
@@ -63,6 +65,10 @@ export default class EnrichmentServiceClearbit extends LoggerBase implements IEn
     return enriched
   }
 
+  async hasRemainingCredits(): Promise<boolean> {
+    return true
+  }
+
   private async getDataUsingEmail(email: string): Promise<IMemberEnrichmentDataClearbit> {
     let response: IMemberEnrichmentClearbitAPIResponse
 

services/apps/premium/members_enrichment_worker/src/sources/progai/service.ts

@@ -124,6 +124,10 @@ export default class EnrichmentServiceProgAI extends LoggerBase implements IEnri
     return enrichableUsingGithubHandle || enrichableUsingEmail
   }
 
+  async hasRemainingCredits(): Promise<boolean> {
+    return true
+  }
+
   async getData(input: IEnrichmentSourceInput): Promise<IMemberEnrichmentDataProgAI> {
     let enriched: IMemberEnrichmentDataProgAI = null
 

services/apps/premium/members_enrichment_worker/src/sources/serp/service.ts

@@ -9,12 +9,16 @@ import {
   IMemberEnrichmentDataNormalized,
 } from '../../types'
 
-import { IMemberEnrichmentDataSerp, IMemberEnrichmentSerpApiResponse } from './types'
+import {
+  IMemberEnrichmentDataSerp,
+  IMemberEnrichmentSerpApiResponse,
+  ISerpApiAccountUsageData,
+} from './types'
 
 export default class EnrichmentServiceSerpApi extends LoggerBase implements IEnrichmentService {
   public source: MemberEnrichmentSource = MemberEnrichmentSource.SERP
   public platform = `enrichment-${this.source}`
-  public enrichMembersWithActivityMoreThan = 10
+  public enrichMembersWithActivityMoreThan = 500
 
   public enrichableBySql = `
   ("activitySummary".total_count > ${this.enrichMembersWithActivityMoreThan}) AND
@@ -45,6 +49,25 @@ export default class EnrichmentServiceSerpApi extends LoggerBase implements IEnr
     )
   }
 
+  async hasRemainingCredits(): Promise<boolean> {
+    try {
+      const config = {
+        method: 'get',
+        url: `https://serpapi.com/account`,
+        params: {
+          api_key: process.env['CROWD_ENRICHMENT_SERP_API_KEY'],
+        },
+      }
+
+      const response: ISerpApiAccountUsageData = (await axios(config)).data
+
+      return response.total_searches_left > 0
+    } catch (error) {
+      this.log.error('Error while checking serpapi account usage', error)
+      return false
+    }
+  }
+
   async getData(input: IEnrichmentSourceInput): Promise<IMemberEnrichmentDataSerp | null> {
     let enriched: IMemberEnrichmentDataSerp = null
 

services/apps/premium/members_enrichment_worker/src/sources/serp/types.ts

@@ -7,6 +7,24 @@ export interface IMemberEnrichmentSerpApiResponse {
   search_information: IMemberEnrichmentSerpApiResponseSearchInformation
 }
 
+export interface ISerpApiAccountUsageData {
+  account_id: string
+  api_key: string
+  account_email: string
+  account_status: string
+  plan_id: string
+  plan_name: string
+  plan_monthly_price: number
+  searches_per_month: number
+  plan_searches_left: number
+  extra_credits: number
+  total_searches_left: number
+  this_month_usage: number
+  this_hour_searches: number
+  last_hour_searches: number
+  account_rate_limit_per_hour: number
+}
+
 export interface IMemberEnrichmentSerpApiResponseSearchInformation {
   query_displayed: string
   total_results: number

services/apps/premium/members_enrichment_worker/src/types.ts

@@ -37,6 +37,10 @@ export interface IEnrichmentService {
   // can the source enrich using this input
   isEnrichableBySource(input: IEnrichmentSourceInput): boolean
 
+  // does the source have credits to enrich members, if returned false the source will be skipped
+  // response will be saved to redis for 60 seconds and will be used for subsequent calls
+  hasRemainingCredits(): Promise<boolean>
+
   // SQL filter to get enrichable members for a source
   // members table is available as "members" alias
   // memberIdentities table is available as "mi" alias
@@ -66,7 +70,7 @@ export interface IMemberEnrichmentDataNormalizedOrganization {
 }
 
 export interface IGetMembersForEnrichmentArgs {
-  afterId?: string
+  afterCursor: { activityCount: number; memberId: string } | null
 }
 
 export interface IMemberEnrichmentSocialData {

services/apps/premium/members_enrichment_worker/src/workflows/enrichMember.ts

@@ -19,6 +19,7 @@ const {
   updateMemberEnrichmentCache,
   isCacheObsolete,
   normalizeEnrichmentData,
+  findMemberIdentityWithTheMostActivityInPlatform,
 } = proxyActivities<typeof activities>({
   startToCloseTimeout: '20 seconds',
   retry: {
@@ -42,12 +43,6 @@ export async function enrichMember(
     // cache is obsolete when it's not found or cache.updatedAt is older than cacheObsoleteAfterSeconds
     if (await isCacheObsolete(source, cache)) {
       const enrichmentInput: IEnrichmentSourceInput = {
-        github: input.identities.find(
-          (i) =>
-            i.verified &&
-            i.platform === PlatformType.GITHUB &&
-            i.type === MemberIdentityType.USERNAME,
-        ),
         email: input.identities.find((i) => i.verified && i.type === MemberIdentityType.EMAIL),
         linkedin: input.identities.find(
           (i) =>
@@ -61,6 +56,30 @@ export async function enrichMember(
         activityCount: input.activityCount || 0,
       }
 
+      // there can be multiple verified identities in github, we select the one with the most activities
+      const verifiedGithubIdentities = input.identities.filter(
+        (i) =>
+          i.verified &&
+          i.platform === PlatformType.GITHUB &&
+          i.type === MemberIdentityType.USERNAME,
+      )
+
+      if (verifiedGithubIdentities.length > 1) {
+        const ghIdentityWithTheMostActivities =
+          await findMemberIdentityWithTheMostActivityInPlatform(input.id, PlatformType.GITHUB)
+        if (ghIdentityWithTheMostActivities) {
+          enrichmentInput.github = input.identities.find(
+            (i) =>
+              i.verified &&
+              i.platform === PlatformType.GITHUB &&
+              i.type === MemberIdentityType.USERNAME &&
+              i.value === ghIdentityWithTheMostActivities.username,
+          )
+        }
+      } else {
+        enrichmentInput.github = verifiedGithubIdentities?.[0] || undefined
+      }
+
       const data = await getEnrichmentData(source, enrichmentInput)
 
       if (!cache) {