feat: evaluation priority logic (CM-1166) (#4130)

Signed-off-by: Umberto Sgueglia <usgueglia@contractor.linuxfoundation.org>

Author: ulemons

services/apps/automatic_projects_discovery_worker/src/sources/insights-discussions/source.ts

@@ -261,7 +261,6 @@ export class InsightsDiscussionsSource implements IDiscoverySource {
       projectSlug,
       repoName,
       repoUrl,
-      action: 'evaluate',
     }
   }
 }

services/apps/projects_evaluation_worker/src/activities/activities.ts

@@ -1,13 +1,34 @@
-import { findProjectCatalogPendingEvaluation, updateProjectCatalog } from '@crowd/data-access-layer'
+import {
+  findProjectCatalogPendingEvaluation,
+  promoteProjectsToEvaluate,
+  updateProjectCatalog,
+} from '@crowd/data-access-layer'
 import { IDbProjectCatalog } from '@crowd/data-access-layer/src/project-catalog/types'
 import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
 import { getServiceLogger } from '@crowd/logging'
 
 import { evaluateProject } from '../evaluator/evaluator'
 import { svc } from '../main'
+import { IPriorityConfig } from '../types'
 
 const log = getServiceLogger()
 
+/**
+ * Promotes 'auto' projects to 'evaluate' up to the configured limit.
+ * Count, slot computation, locking, and update are all done atomically
+ * inside a single SQL statement — see promoteProjectsToEvaluate in the DAL.
+ */
+export async function promoteProjectsForEvaluation(config: IPriorityConfig): Promise<void> {
+  const { evaluateLimit, sourcePriority } = config
+  const qx = pgpQx(svc.postgres.writer.connection())
+
+  log.info({ evaluateLimit, sourcePriority }, 'Priority promotion: starting.')
+
+  const promoted = await promoteProjectsToEvaluate(qx, { evaluateLimit, sourcePriority })
+
+  log.info({ promoted }, 'Priority promotion: complete.')
+}
+
 export async function fetchPendingProjects(batchSize: number): Promise<IDbProjectCatalog[]> {
   const qx = pgpQx(svc.postgres.reader.connection())
 

services/apps/projects_evaluation_worker/src/schedules/scheduleProjectsEvaluation.ts

@@ -1,7 +1,18 @@
 import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client'
 
 import { svc } from '../main'
-import { evaluateProjects } from '../workflows'
+import { IEvaluateProjectsInput, evaluateProjects } from '../workflows'
+
+// Priority configuration for the evaluation queue.
+// - evaluateLimit: maximum number of projects in 'evaluate' state at any time.
+// - sourcePriority: ordered list of sources; earlier = higher priority; unlisted sources rank last.
+const EVALUATION_ARGS: IEvaluateProjectsInput = {
+  batchSize: 50,
+  priorityConfig: {
+    evaluateLimit: 50,
+    sourcePriority: ['insights-discussions'],
+  },
+}
 
 export const scheduleProjectsEvaluation = async () => {
   svc.log.info('Scheduling projects evaluation')
@@ -21,8 +32,8 @@ export const scheduleProjectsEvaluation = async () => {
         type: 'startWorkflow',
         workflowType: evaluateProjects,
         taskQueue: 'projects-evaluation',
-        args: [{ batchSize: 100 }],
-        // 100 projects × ~3min each = ~5h worst case; set ceiling with margin.
+        args: [EVALUATION_ARGS],
+        // 50 projects × ~3min each = ~2.5h worst case; set ceiling with margin.
         workflowExecutionTimeout: '6 hours',
         retry: {
           initialInterval: '30 seconds',

services/apps/projects_evaluation_worker/src/types.ts

@@ -0,0 +1,14 @@
+export interface IPriorityConfig {
+  /** Maximum number of projects in the 'evaluate' queue at any time. */
+  evaluateLimit: number
+  /**
+   * Ordered list of source names by descending priority.
+   * Sources not in this list rank below all listed ones.
+   */
+  sourcePriority: string[]
+}
+
+export interface IEvaluateProjectsInput {
+  batchSize?: number
+  priorityConfig?: IPriorityConfig
+}

services/apps/projects_evaluation_worker/src/workflows.ts

@@ -1,3 +1,5 @@
+import type { IEvaluateProjectsInput } from './types'
 import { evaluateProjects } from './workflows/evaluateProjects'
 
 export { evaluateProjects }
+export type { IEvaluateProjectsInput }

services/apps/projects_evaluation_worker/src/workflows/evaluateProjects.ts

@@ -1,6 +1,13 @@
 import { log, proxyActivities } from '@temporalio/workflow'
 
 import type * as activities from '../activities'
+import type { IEvaluateProjectsInput, IPriorityConfig } from '../types'
+
+// Quick DB write — promote auto → evaluate.
+const promotionActivities = proxyActivities<typeof activities>({
+  startToCloseTimeout: '1 minute',
+  retry: { maximumAttempts: 3 },
+})
 
 // Short timeout: just a DB read.
 const fetchActivities = proxyActivities<typeof activities>({
@@ -14,13 +21,20 @@ const evaluateActivities = proxyActivities<typeof activities>({
   retry: { maximumAttempts: 2 },
 })
 
-const DEFAULT_BATCH_SIZE = 100
+const DEFAULT_PRIORITY_CONFIG: IPriorityConfig = {
+  evaluateLimit: 50,
+  sourcePriority: ['insights-discussions'],
+}
 
-export async function evaluateProjects(input: { batchSize?: number } = {}): Promise<void> {
-  const batchSize = input.batchSize ?? DEFAULT_BATCH_SIZE
+export async function evaluateProjects(input: IEvaluateProjectsInput = {}): Promise<void> {
+  const { batchSize = 50, priorityConfig = DEFAULT_PRIORITY_CONFIG } = input
 
   log.info('evaluateProjects workflow started.')
 
+  // Step 1: promote 'auto' projects to 'evaluate' according to priority config.
+  await promotionActivities.promoteProjectsForEvaluation(priorityConfig)
+
+  // Step 2: fetch the evaluation queue (includes any leftovers from prior runs).
   const projects = await fetchActivities.fetchPendingProjects(batchSize)
 
   if (projects.length === 0) {

services/libs/data-access-layer/src/project-catalog/projectCatalog.ts

@@ -1,7 +1,12 @@
 import { QueryExecutor } from '../queryExecutor'
 import { prepareSelectColumns } from '../utils'
 
-import { IDbProjectCatalog, IDbProjectCatalogCreate, IDbProjectCatalogUpdate } from './types'
+import {
+  IDbProjectCatalog,
+  IDbProjectCatalogCreate,
+  IDbProjectCatalogUpdate,
+  ProjectCatalogAction,
+} from './types'
 
 const PROJECT_CATALOG_COLUMNS = [
   'id',
@@ -108,6 +113,79 @@ export async function countProjectCatalog(qx: QueryExecutor): Promise<number> {
   return parseInt(result.count, 10)
 }
 
+export async function countProjectCatalogByAction(
+  qx: QueryExecutor,
+  action: ProjectCatalogAction,
+): Promise<number> {
+  const result = await qx.selectOne(
+    `
+    SELECT COUNT(*) AS count
+    FROM "projectCatalog"
+    WHERE action = $(action)
+    `,
+    { action },
+  )
+  return parseInt(result.count, 10)
+}
+
+/**
+ * Promotes 'auto' projects to 'evaluate', targeting a soft cap on the queue size.
+ *
+ * Uses a CTE to:
+ *   1. Compute available slots (evaluateLimit − current 'evaluate' count) inline.
+ *   2. Lock candidates with FOR UPDATE SKIP LOCKED — prevents double-promotion
+ *      if two transactions run concurrently (e.g. simultaneous manual triggers).
+ *      Note: concurrent calls can each compute the same slot count and promote
+ *      up to evaluateLimit disjoint rows, so the cap is soft — the queue can
+ *      overshoot by at most evaluateLimit per extra concurrent caller. In practice
+ *      this doesn't happen: the schedule uses ScheduleOverlapPolicy.SKIP.
+ *   3. Re-check action = 'auto' in the outer UPDATE to guard against rows whose
+ *      state changed after the subquery snapshot (e.g. manual updates).
+ *
+ * Ordering (configurable via `sourcePriority`):
+ *   1. Source priority — earlier position in the array = higher priority (unlisted = lowest)
+ *   2. lfCriticalityScore DESC (NULLs last)
+ *   3. createdAt ASC (stable tie-breaker)
+ *
+ * Returns the number of rows actually promoted.
+ */
+export async function promoteProjectsToEvaluate(
+  qx: QueryExecutor,
+  options: { evaluateLimit: number; sourcePriority: string[] },
+): Promise<number> {
+  const { evaluateLimit, sourcePriority } = options
+
+  return qx.result(
+    `
+    WITH
+      slots AS (
+        SELECT GREATEST(0, $(evaluateLimit) - COUNT(*)) AS available
+        FROM "projectCatalog"
+        WHERE action = 'evaluate'
+      ),
+      candidates AS (
+        SELECT pc.id
+        FROM "projectCatalog" pc
+        CROSS JOIN slots
+        WHERE pc.action = 'auto'
+          AND slots.available > 0
+        ORDER BY
+          COALESCE(ARRAY_POSITION($(sourcePriority)::text[], pc.source), 2147483647) ASC,
+          pc."lfCriticalityScore" DESC NULLS LAST,
+          pc."createdAt" ASC
+        LIMIT (SELECT available FROM slots)
+        FOR UPDATE SKIP LOCKED
+      )
+    UPDATE "projectCatalog"
+    SET action = 'evaluate', "updatedAt" = NOW()
+    FROM candidates
+    WHERE "projectCatalog".id = candidates.id
+      AND "projectCatalog".action = 'auto'
+    `,
+    { evaluateLimit, sourcePriority },
+  )
+}
+
 export async function insertProjectCatalog(
   qx: QueryExecutor,
   data: IDbProjectCatalogCreate,