fix: harden mailing list bug fixes and add activity type wiring (CM-1318)

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

Author: themarolt

backend/package.json

@@ -33,7 +33,11 @@
     "script:fix-duplicate-members": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/fix-duplicate-members.ts",
     "script:fix-members-activities-after-unaffilation": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/fix-members-activities-after-unaffilation.ts",
     "script:process-bot-members": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/process-bot-members.ts",
-    "script:backfill-email-domain-member-organization-dates": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/backfill-email-domain-member-organization-dates.ts"
+    "script:backfill-email-domain-member-organization-dates": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/backfill-email-domain-member-organization-dates.ts",
+    "script:onboard-default-tenant": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/onboard-default-tenant.ts",
+    "script:onboard-default-tenant:local": "set -a && . ./.env.dist.local && . ./.env.override.local && set +a && pnpm run script:onboard-default-tenant",
+    "script:create-mailing-list-integration": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/create-mailing-list-integration.ts",
+    "script:create-mailing-list-integration:local": "set -a && . ./.env.dist.local && . ./.env.override.local && set +a && SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/create-mailing-list-integration.ts"
   },
   "lint-staged": {
     "**/*.ts": [

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

@@ -1,13 +1,24 @@
+import { z } from 'zod'
+
 import Permissions from '../../../security/permissions'
 import IntegrationService from '../../../services/integrationService'
 import PermissionChecker from '../../../services/user/permissionChecker'
+import { validateOrThrow } from '../../../utils/validation'
+
+const bodySchema = z.object({
+  lists: z
+    .array(
+      z.object({
+        name: z.string().trim().min(1),
+        sourceUrl: z.string().trim().min(1),
+      }),
+    )
+    .default([]),
+})
 
 export default async (req, res) => {
   new PermissionChecker(req).validateHas(Permissions.values.tenantEdit)
-  const integrationData = {
-    ...req.body,
-    lists: req.body.lists || [],
-  }
+  const integrationData = validateOrThrow(bodySchema, req.body)
 
   const payload = await new IntegrationService(req).mailingListConnectOrUpdate(integrationData)
   await req.responseHandler.success(req, res, payload)

backend/src/bin/scripts/create-mailing-list-integration.ts

@@ -0,0 +1,110 @@
+/* eslint-disable no-console */
+
+/* eslint-disable import/no-extraneous-dependencies */
+import commandLineArgs from 'command-line-args'
+import commandLineUsage from 'command-line-usage'
+import * as fs from 'fs'
+
+import { DEFAULT_TENANT_ID, generateUUIDv1 } from '@crowd/common'
+
+import SegmentRepository from '@/database/repositories/segmentRepository'
+import SequelizeRepository from '@/database/repositories/sequelizeRepository'
+import IntegrationService from '@/services/integrationService'
+
+const options = [
+  {
+    name: 'help',
+    alias: 'h',
+    type: Boolean,
+    description: 'Print this usage guide.',
+  },
+  {
+    name: 'file',
+    alias: 'f',
+    type: String,
+    description:
+      'Path to a JSON file with the lists array, each entry shaped as name + sourceUrl ' +
+      '(e.g. name "linux-serial", sourceUrl https://lore.kernel.org/linux-serial).',
+  },
+  {
+    name: 'segment',
+    alias: 's',
+    type: String,
+    description: "Segment id. Optional — defaults to the tenant's first subproject.",
+  },
+]
+
+const sections = [
+  {
+    header: 'Create mailing list integration',
+    content:
+      'Calls IntegrationService.mailingListConnectOrUpdate() directly to create/update the ' +
+      'mailinglist integration and onboard its lists for processing, until the frontend connect ' +
+      'UI ships (CM-1318).',
+  },
+  {
+    header: 'Options',
+    optionList: options,
+  },
+]
+
+const usage = commandLineUsage(sections)
+// pnpm forwards a literal `--` separator when args are passed via `pnpm run ... -- <args>`;
+// strip it so command-line-args doesn't choke on it.
+const argv = process.argv.slice(2).filter((arg) => arg !== '--')
+const parameters = commandLineArgs(options, { argv })
+
+if (parameters.help || !parameters.file) {
+  console.log(usage)
+  process.exit(parameters.help ? 0 : 1)
+} else {
+  setImmediate(async () => {
+    let fileContents: string
+    try {
+      fileContents = fs.readFileSync(parameters.file, 'utf-8')
+    } catch (err) {
+      console.error(`Could not read file at ${parameters.file}: ${(err as Error).message}`)
+      process.exit(1)
+    }
+
+    let lists
+    try {
+      lists = JSON.parse(fileContents)
+    } catch (err) {
+      console.error(`File at ${parameters.file} is not valid JSON: ${(err as Error).message}`)
+      process.exit(1)
+    }
+
+    const repoOptions = await SequelizeRepository.getDefaultIRepositoryOptions()
+    repoOptions.currentTenant = { id: DEFAULT_TENANT_ID }
+    ;(repoOptions as unknown as { requestId: string }).requestId = generateUUIDv1()
+
+    const adminUser = await repoOptions.database.user.findOne({ where: {} })
+    if (!adminUser) {
+      console.error('No user found — run script:onboard-default-tenant first.')
+      process.exit(1)
+    }
+    repoOptions.currentUser = adminUser
+
+    const segmentRepository = new SegmentRepository(repoOptions)
+    repoOptions.currentSegments = parameters.segment
+      ? await segmentRepository.findInIds([parameters.segment])
+      : (await segmentRepository.querySubprojects({ limit: 1, offset: 0 })).rows
+
+    if (repoOptions.currentSegments.length === 0) {
+      console.error('No segment found/resolved — pass --segment explicitly.')
+      process.exit(1)
+    }
+
+    console.log(`Segment: ${repoOptions.currentSegments[0].id} (${repoOptions.currentSegments[0].name})`)
+    console.log('Lists:', JSON.stringify(lists, null, 2))
+
+    const integration = await new IntegrationService(repoOptions).mailingListConnectOrUpdate(
+      { lists },
+      repoOptions,
+    )
+
+    console.log('Integration:', JSON.stringify(integration, null, 2))
+    process.exit(0)
+  })
+}

backend/src/bin/scripts/onboard-default-tenant.ts

@@ -0,0 +1,74 @@
+/* eslint-disable no-console */
+
+/* eslint-disable import/no-extraneous-dependencies */
+import commandLineArgs from 'command-line-args'
+import commandLineUsage from 'command-line-usage'
+
+import SequelizeRepository from '@/database/repositories/sequelizeRepository'
+import AuthService from '@/services/auth/authService'
+
+const options = [
+  {
+    name: 'help',
+    alias: 'h',
+    type: Boolean,
+    description: 'Print this usage guide.',
+  },
+  {
+    name: 'email',
+    alias: 'e',
+    type: String,
+    defaultValue: 'local-dev@example.com',
+    description: 'Email for the dev user driving the onboarding (default: local-dev@example.com)',
+  },
+]
+
+const sections = [
+  {
+    header: 'Onboard default tenant',
+    content:
+      'Runs the same onboarding logic as a real Auth0 signup (AuthService.signinFromSSO -> ' +
+      'handleOnboard -> TenantService.createOrJoinDefault), without needing a real Auth0 token. ' +
+      'Creates the default tenant, default segment, settings, and an admin user/tenantUser row. ' +
+      'Safe to re-run — createOrJoinDefault joins the existing tenant instead of duplicating it.',
+  },
+  {
+    header: 'Options',
+    optionList: options,
+  },
+]
+
+const usage = commandLineUsage(sections)
+const parameters = commandLineArgs(options)
+
+if (parameters.help) {
+  console.log(usage)
+} else {
+  setImmediate(async () => {
+    const repoOptions = await SequelizeRepository.getDefaultIRepositoryOptions()
+
+    await AuthService.signinFromSSO(
+      'auth0',
+      `dev|${parameters.email}`,
+      parameters.email,
+      true,
+      'Local',
+      'Dev',
+      'Local Dev',
+      null,
+      null,
+      null,
+      repoOptions,
+    )
+
+    const tenant = await repoOptions.database.tenant.findOne({ where: {} })
+    const segment = await repoOptions.database.segment.findOne({
+      where: { tenantId: tenant.id },
+    })
+
+    console.log(`Tenant: ${tenant.id} (${tenant.name})`)
+    console.log(`Segment: ${segment.id} (${segment.name})`)
+
+    process.exit(0)
+  })
+}

backend/src/database/migrations/V1784291394__add_mailinglist_activity_type.sql

@@ -0,0 +1,3 @@
+-- Mailing list (Collaboration)
+INSERT INTO "activityTypes" ("activityType", platform, "isCodeContribution", "isCollaboration", description) VALUES
+('message', 'mailinglist', false, true, 'Sent a message to a mailing list');

docs/adr/0009-mailinglist-skip-unparseable-dates.md

@@ -0,0 +1,67 @@
+# ADR-0009: Skip Mailing List Activities With Unparseable/Implausible Dates
+
+**Date**: 2026-07-19
+**Status**: accepted
+**Deciders**: Uros Marolt
+
+## Context
+
+The mailing list integration (`services/apps/mailing_list_integration`) parses
+RFC2822 `Date` headers from archived mailing list messages
+(`crowdmail/services/parse/noteren.py`, `parse_email()`) to build each
+activity's `timestamp`. During testing against a real archive
+(`linux-serial`, mirrored from lore.kernel.org), a message surfaced whose
+`Date` header parsed "successfully" via `email.utils.parsedate_tz()` but
+produced an implausible year (e.g. `102` instead of a 4-digit year).
+Python's stdlib already RFC2822-normalizes legitimate 2-digit years (`68` →
+`2068`/`02` → `2002`), so this is not a parsing bug in our port — the
+`Date` header itself is malformed in the source archive. Without a guard,
+`mktime_tz()` + `strftime("%Y-...")` emits a non-4-digit year (e.g.
+`"102-06-12T11:56:20.000000Z"`), which downstream JS `Date` parsing (in
+the Node ingestion pipeline) rejects, causing per-message ingestion errors.
+
+## Decision
+
+Reject any parsed date whose year falls outside `1970–9999` by raising
+`ValueError` (caught by the existing `except (ValueError, OverflowError)`
+block), leaving `timestamp` empty for that message. `list_worker.py` skips
+queuing any activity with an empty timestamp. The message is not ingested
+as an activity, and its shard head still advances past it (so it is not
+retried on subsequent polls).
+
+## Alternatives Considered
+
+### Alternative 1: Heuristically repair malformed years
+- **Pros**: Could recover activities that would otherwise be dropped.
+- **Cons**: No reliable signal for what year was intended once the header
+  is already malformed at the source (e.g. `102` — off by 1900? 2000?
+  truncated?).
+- **Why not**: Any heuristic risks silently mis-dating an activity, which
+  is worse than dropping it — a wrong timestamp corrupts activity
+  timelines and trend analytics without any visible error.
+
+### Alternative 2: Ingest with the corrupted timestamp as-is
+- **Pros**: No data loss.
+- **Cons**: Breaks downstream JS `Date` parsing in the ingestion pipeline,
+  causing hard failures rather than a clean skip.
+- **Why not**: A hard pipeline failure for one bad message is worse than
+  skipping that one message.
+
+## Consequences
+
+### Positive
+- Malformed dates never reach the ingestion pipeline as corrupt
+  timestamps; no pipeline-level failures from a single bad message.
+- Matches the existing skip pattern already used for other
+  unparseable/missing fields in this parser.
+
+### Negative
+- Messages with malformed `Date` headers are silently and permanently
+  dropped — never retried, since the shard head advances past them
+  regardless. The only visible trace is a `logging.warning()` line.
+
+### Risks
+- If a source archive has a systemic date-formatting issue (not just rare
+  garbage), this could silently drop a meaningful fraction of activities.
+  Mitigation: the warning log includes the raw header and commit id, so a
+  spike can be diagnosed via log volume if ever suspected.

docs/adr/README.md

@@ -15,6 +15,7 @@ Use the `/adr` skill in Claude Code to record new ADRs or query past decisions.
 | [ADR-0006](./0006-database-schema-types-as-source-of-truth.md) | Database schema types as the source of truth | accepted | 2026-07-09 |
 | [ADR-0007](./0007-test-factory-primitives-and-defaults.md) | Test factory primitives and defaults | accepted | 2026-07-10 |
 | [ADR-0008](./0008-how-we-write-unit-tests.md) | How we write unit tests | accepted | 2026-07-13 |
+| [ADR-0009](./0009-mailinglist-skip-unparseable-dates.md) | Skip mailing list activities with unparseable/implausible dates | accepted | 2026-07-19 |
 
 ## Why ADRs?
 

scripts/services/docker/Dockerfile.mailing_list_integration

@@ -41,8 +41,10 @@ RUN --mount=type=cache,target=/root/.cache/uv \
 FROM base AS runner
 
 # Install runtime dependencies, including public-inbox for mirroring lists
+# curl is required by public-inbox-clone itself (not pulled in via --no-install-recommends)
 RUN apt-get update && apt-get install -y \
     ca-certificates \
+    curl \
     git \
     public-inbox \
     --no-install-recommends \

services/apps/data_sink_worker/src/service/activity.service.ts

@@ -1162,7 +1162,7 @@ export default class ActivityService extends LoggerBase {
 
     // find members to create
     for (const payload of payloadsWithoutDbMembers) {
-      const key = `${payload.platform}:${payload.activity.username}`
+      const key = `${payload.platform}:${payload.activity.username?.toLowerCase()}`
       if (!membersToCreateMap.has(key)) {
         const segmentIds = new Set<string>()
         segmentIds.add(payload.segmentId)
@@ -1188,7 +1188,7 @@ export default class ActivityService extends LoggerBase {
 
     // find object members to create
     for (const payload of payloadsWithoutDbObjectMembers) {
-      const key = `${payload.platform}:${payload.activity.objectMemberUsername}`
+      const key = `${payload.platform}:${payload.activity.objectMemberUsername?.toLowerCase()}`
       if (!membersToCreateMap.has(key)) {
         const segmentIds = new Set<string>()
         segmentIds.add(payload.segmentId)
@@ -1756,12 +1756,17 @@ export default class ActivityService extends LoggerBase {
     memberType: 'member' | 'objectMember',
     dbMember?: IDbMember,
   ): Promise<string | Record<string, unknown> | undefined> {
+    const IDENTITY_CONSTRAINTS = new Set([
+      'uix_memberIdentities_platform_value_type_verified',
+      'uix_memberIdentities_memberId_platform_value_type',
+    ])
+
     const checkForIdentityConstraint = (error: any): boolean => {
       if (
         error.constructor &&
         error.constructor.name === 'DatabaseError' &&
         error.constraint &&
-        error.constraint === 'uix_memberIdentities_platform_value_type_verified'
+        IDENTITY_CONSTRAINTS.has(error.constraint)
       ) {
         return true
       }
@@ -1814,6 +1819,18 @@ export default class ActivityService extends LoggerBase {
         }
       }
 
+      // Create path: no dbMember to compare against yet (member row doesn't exist locally).
+      // The verified-identity index is globally unique, so a found owner IS the canonical
+      // member — attach to it instead of failing (covers whole-batch-retry re-attempting a
+      // create whose identity insert already succeeded in an earlier attempt).
+      if (ownerId && !dbMember) {
+        this.log.warn(
+          { memberId: ownerId, identity: conflictIdentity },
+          'Verified identity already belongs to an existing member — attaching to it instead of creating a new one',
+        )
+        return ownerId
+      }
+
       // Pass 2: if no external conflict, check whether this member already owns the
       // identity (stale-prefetch race). Re-uses the owners map — no extra DB query.
       if (!ownerId && dbMember) {