feat: add mailing list onboarding endpoint (CM-1318)

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

backend/src/api/integration/helpers/mailingListAuthenticate.ts

@@ -0,0 +1,14 @@
+import Permissions from '../../../security/permissions'
+import IntegrationService from '../../../services/integrationService'
+import PermissionChecker from '../../../services/user/permissionChecker'
+
+export default async (req, res) => {
+  new PermissionChecker(req).validateHas(Permissions.values.tenantEdit)
+  const integrationData = {
+    ...req.body,
+    lists: req.body.lists || [],
+  }
+
+  const payload = await new IntegrationService(req).mailingListConnectOrUpdate(integrationData)
+  await req.responseHandler.success(req, res, payload)
+}

backend/src/api/integration/index.ts

@@ -77,6 +77,8 @@ export default (app) => {
 
   // Git
   app.put(`/git-connect`, safeWrap(require('./helpers/gitAuthenticate').default))
+
+  app.put(`/mailing-list-connect`, safeWrap(require('./helpers/mailingListAuthenticate').default))
   app.put(`/confluence-connect`, safeWrap(require('./helpers/confluenceAuthenticate').default))
   app.put(`/gerrit-connect`, safeWrap(require('./helpers/gerritAuthenticate').default))
   app.get('/devto-validate', safeWrap(require('./helpers/devtoValidators').default))

backend/src/services/integrationService.ts

@@ -16,6 +16,7 @@ import {
 } from '@crowd/common'
 import { CommonIntegrationService, getGithubInstallationToken } from '@crowd/common_services'
 import { ICreateInsightsProject } from '@crowd/data-access-layer/src/collections'
+import { upsertMailingLists } from '@crowd/data-access-layer/src/mailinglist'
 import {
   ICreateRepository,
   IRepository,
@@ -1423,6 +1424,61 @@ export default class IntegrationService {
     return integration
   }
 
+  /**
+   * Adds/updates a mailing list (public-inbox/lore) integration and onboards
+   * its lists for processing by the mailing_list_integration worker.
+   *
+   * @param integrationData.lists - Mailing lists to onboard (name + sourceUrl)
+   * @param options - Optional repository options
+   */
+  async mailingListConnectOrUpdate(
+    integrationData: {
+      lists: Array<{ name: string; sourceUrl: string }>
+    },
+    options?: IRepositoryOptions,
+  ) {
+    const lists = integrationData.lists || []
+
+    if (lists.length === 0) {
+      this.options.log.warn('No lists provided - skipping mailing list integration update')
+      return null
+    }
+
+    const currentOptions = options || this.options
+    const existingTransaction =
+      currentOptions.transaction || SequelizeRepository.getTransaction(currentOptions)
+    const transaction =
+      existingTransaction || (await SequelizeRepository.createTransaction(options || this.options))
+    let integration
+
+    try {
+      integration = await this.createOrUpdate(
+        {
+          platform: PlatformType.MAILINGLIST,
+          settings: { lists },
+          status: 'done',
+        },
+        transaction,
+        options,
+      )
+
+      const currentSegmentId = (options || this.options).currentSegments[0].id
+      const qx = SequelizeRepository.getQueryExecutor({ ...(options || this.options), transaction })
+      await upsertMailingLists(qx, currentSegmentId, integration.id, lists)
+
+      if (!existingTransaction) {
+        await SequelizeRepository.commitTransaction(transaction)
+      }
+    } catch (err) {
+      if (!existingTransaction) {
+        await SequelizeRepository.rollbackTransaction(transaction)
+      }
+      this.options.log.error(`mailingListConnectOrUpdate failed with error: ${err}`)
+      throw err
+    }
+    return integration
+  }
+
   async atlassianAdminConnect(adminApi: string, organizationId: string) {
     const nangoPayload = {
       params: {

services/libs/data-access-layer/src/mailinglist/index.ts

@@ -0,0 +1,57 @@
+import { QueryExecutor } from '../queryExecutor'
+
+export interface IMailingListToOnboard {
+  name: string
+  sourceUrl: string
+}
+
+/**
+ * Upsert mailing lists (public-inbox/lore) for a segment/integration and
+ * seed their processing state so the mailing_list_integration worker picks
+ * them up. Re-running with the same sourceUrl re-points the list at the
+ * given segment/integration without resetting its processing progress.
+ * @param qx - Query executor (should be transactional)
+ * @param segmentId - Segment the lists belong to
+ * @param integrationId - Integration these lists are onboarded under
+ * @param lists - Mailing lists to onboard, keyed by sourceUrl
+ */
+export async function upsertMailingLists(
+  qx: QueryExecutor,
+  segmentId: string,
+  integrationId: string,
+  lists: IMailingListToOnboard[],
+): Promise<string[]> {
+  if (lists.length === 0) {
+    return []
+  }
+
+  const rows = await qx.select(
+    `
+    WITH upserted_list AS (
+      INSERT INTO mailinglist.lists (id, name, "sourceUrl", "segmentId", "integrationId", "createdAt", "updatedAt")
+      SELECT uuid_generate_v4(), v.name, v."sourceUrl", $(segmentId)::uuid, $(integrationId)::uuid, NOW(), NOW()
+      FROM json_to_recordset($(lists)::json) AS v(name text, "sourceUrl" text)
+      ON CONFLICT ("sourceUrl") DO UPDATE SET
+        name = EXCLUDED.name,
+        "segmentId" = EXCLUDED."segmentId",
+        "integrationId" = EXCLUDED."integrationId",
+        "updatedAt" = NOW()
+      RETURNING id
+    ),
+    seed_processing AS (
+      INSERT INTO mailinglist."listProcessing" ("listId", state, priority, "lastProcessedHeads", "createdAt", "updatedAt")
+      SELECT id, 'pending', 2, '{}'::jsonb, NOW(), NOW()
+      FROM upserted_list
+      ON CONFLICT ("listId") DO NOTHING
+    )
+    SELECT id FROM upserted_list
+    `,
+    {
+      segmentId,
+      integrationId,
+      lists: JSON.stringify(lists),
+    },
+  )
+
+  return rows.map((row: { id: string }) => row.id)
+}

services/libs/types/src/enums/platforms.ts

@@ -20,6 +20,7 @@ export enum PlatformType {
   GIT = 'git',
   CRUNCHBASE = 'crunchbase',
   GROUPSIO = 'groupsio',
+  MAILINGLIST = 'mailinglist',
   COMMITTEES = 'committees',
   CONFLUENCE = 'confluence',
   GERRIT = 'gerrit',