diff --git a/backend/src/database/migrations/V1792094459__add_octolens_mention_activity_types.sql b/backend/src/database/migrations/V1792094459__add_octolens_mention_activity_types.sql new file mode 100644 index 0000000000..a05f34d5df --- /dev/null +++ b/backend/src/database/migrations/V1792094459__add_octolens_mention_activity_types.sql @@ -0,0 +1,25 @@ +-- The table has no unique key, so skip rows that already exist (e.g. mention/twitter). +INSERT INTO "activityTypes" ("activityType", platform, "isCodeContribution", "isCollaboration", description, "label") +SELECT v."activityType", v.platform, false, false, v.description, v."label" +FROM (VALUES + ('mention', 'github', 'Mentioned the project on GitHub', 'Mentioned the project'), + ('mention', 'twitter', 'Mentioned the project on X/Twitter', 'Mentioned the project'), + ('mention', 'linkedin', 'Mentioned the project on LinkedIn', 'Mentioned the project'), + ('mention', 'reddit', 'Mentioned the project on Reddit', 'Mentioned the project'), + ('mention', 'devto', 'Mentioned the project on DEV', 'Mentioned the project'), + ('mention', 'hackernews', 'Mentioned the project on Hacker News', 'Mentioned the project'), + ('mention', 'stackoverflow', 'Mentioned the project on Stack Overflow', 'Mentioned the project'), + ('mention', 'medium', 'Mentioned the project on Medium', 'Mentioned the project'), + ('mention', 'producthunt', 'Mentioned the project on Product Hunt', 'Mentioned the project'), + ('mention', 'facebook', 'Mentioned the project on Facebook', 'Mentioned the project'), + ('mention', 'youtube', 'Mentioned the project on YouTube', 'Mentioned the project'), + ('mention', 'bluesky', 'Mentioned the project on Bluesky', 'Mentioned the project'), + ('mention', 'tiktok', 'Mentioned the project on TikTok', 'Mentioned the project'), + ('mention', 'instagram', 'Mentioned the project on Instagram', 'Mentioned the project') +) AS v("activityType", platform, description, "label") +WHERE NOT EXISTS ( + SELECT 1 + FROM "activityTypes" a + WHERE a."activityType" = v."activityType" + AND a.platform = v.platform +); diff --git a/services/apps/data_sink_worker/src/service/activity.service.ts b/services/apps/data_sink_worker/src/service/activity.service.ts index 8792b5d255..f665deff41 100644 --- a/services/apps/data_sink_worker/src/service/activity.service.ts +++ b/services/apps/data_sink_worker/src/service/activity.service.ts @@ -619,7 +619,10 @@ export default class ActivityService extends LoggerBase { continue } - if (payload.platform === PlatformType.GITLAB || payload.platform === PlatformType.GITHUB) { + if ( + !payload.skipRepositorySegmentLookup && + (payload.platform === PlatformType.GITLAB || payload.platform === PlatformType.GITHUB) + ) { repoPayloads.push(payload) } else if (!payload.segmentId) { resultMap.set(payload.resultId, { @@ -2144,6 +2147,7 @@ interface IActivityProcessData { platform: PlatformType activity: IActivityData segmentId: string + skipRepositorySegmentLookup?: boolean dbMember?: IDbMember dbObjectMember?: IDbMember diff --git a/services/apps/data_sink_worker/src/service/dataSink.service.ts b/services/apps/data_sink_worker/src/service/dataSink.service.ts index 9fbff10354..b66bd19c23 100644 --- a/services/apps/data_sink_worker/src/service/dataSink.service.ts +++ b/services/apps/data_sink_worker/src/service/dataSink.service.ts @@ -31,7 +31,10 @@ import { import { WORKER_SETTINGS } from '../conf' import ActivityService from './activity.service' import MemberService from './member.service' -import { resolveActivityResultPlatform } from './resolveActivityResultPlatform' +import { + resolveActivityResultPlatform, + skipsRepositorySegmentLookup, +} from './resolveActivityResultPlatform' /* eslint-disable @typescript-eslint/no-explicit-any */ @@ -351,7 +354,11 @@ export default class DataSinkService extends LoggerBase { if (type === IntegrationResultType.ACTIVITY) { const results = groupedByType.get(type) + const keepsResultSegment = new Set() for (const result of results) { + if (skipsRepositorySegmentLookup(result.platform)) { + keepsResultSegment.add(result.id) + } result.platform = resolveActivityResultPlatform( result.platform, (result.data.data as IActivityData | undefined)?.platform as PlatformType | undefined, @@ -393,6 +400,7 @@ export default class DataSinkService extends LoggerBase { platform: r.platform, activity: r.data.data as IActivityData, segmentId: r.data.segmentId ?? r.segmentId, + skipRepositorySegmentLookup: keepsResultSegment.has(r.id), } }), toProcess.some((r) => r.onboarding === true), diff --git a/services/apps/data_sink_worker/src/service/resolveActivityResultPlatform.test.ts b/services/apps/data_sink_worker/src/service/resolveActivityResultPlatform.test.ts index cc0fc4a3e1..e423ab94ae 100644 --- a/services/apps/data_sink_worker/src/service/resolveActivityResultPlatform.test.ts +++ b/services/apps/data_sink_worker/src/service/resolveActivityResultPlatform.test.ts @@ -2,7 +2,10 @@ import { describe, expect, it } from 'vitest' import { PlatformType } from '@crowd/types' -import { resolveActivityResultPlatform } from './resolveActivityResultPlatform' +import { + resolveActivityResultPlatform, + skipsRepositorySegmentLookup, +} from './resolveActivityResultPlatform' describe('resolveActivityResultPlatform', () => { it('prefers the activity platform over the integration platform when both are set', () => { @@ -29,3 +32,16 @@ describe('resolveActivityResultPlatform', () => { expect(resolveActivityResultPlatform(null, undefined)).toBeNull() }) }) + +describe('skipsRepositorySegmentLookup', () => { + it('skips the repositories lookup for activities from an Octolens integration', () => { + expect(skipsRepositorySegmentLookup(PlatformType.OCTOLENS)).toBe(true) + }) + + it.each([PlatformType.GITHUB, PlatformType.GITHUB_NANGO, PlatformType.GITLAB, null])( + 'keeps the lookup for %s integrations', + (platform) => { + expect(skipsRepositorySegmentLookup(platform)).toBe(false) + }, + ) +}) diff --git a/services/apps/data_sink_worker/src/service/resolveActivityResultPlatform.ts b/services/apps/data_sink_worker/src/service/resolveActivityResultPlatform.ts index 028ea1b69f..8d6e6a2d82 100644 --- a/services/apps/data_sink_worker/src/service/resolveActivityResultPlatform.ts +++ b/services/apps/data_sink_worker/src/service/resolveActivityResultPlatform.ts @@ -9,3 +9,8 @@ export function resolveActivityResultPlatform( const platform = activityPlatform ?? integrationPlatform return platform === PlatformType.GITHUB_NANGO ? PlatformType.GITHUB : platform } + +// Octolens routes each mention to its keyword's project, never through public.repositories. +export function skipsRepositorySegmentLookup(integrationPlatform: PlatformType | null): boolean { + return integrationPlatform === PlatformType.OCTOLENS +} diff --git a/services/apps/organizations_enrichment_worker/src/sources/lfx-internal-api/service.ts b/services/apps/organizations_enrichment_worker/src/sources/lfx-internal-api/service.ts index a793b90b63..1ca09094c2 100644 --- a/services/apps/organizations_enrichment_worker/src/sources/lfx-internal-api/service.ts +++ b/services/apps/organizations_enrichment_worker/src/sources/lfx-internal-api/service.ts @@ -20,6 +20,13 @@ import { IOrganizationEnrichmentDataInternalAPIResponse, } from './types' +// Never stored before these PlatformType values existed; storing them needs its own decision. +const IGNORED_IDENTITY_PLATFORMS = new Set([ + PlatformType.BLUESKY, + PlatformType.TIKTOK, + PlatformType.INSTAGRAM, +]) + export default class EnrichmentServiceLFXInternalAPI extends LoggerBase implements IOrganizationEnrichmentService @@ -126,6 +133,10 @@ export default class EnrichmentServiceLFXInternalAPI this.log.warn({ key }, 'Unknown platform in identities, skipping') return [] } + if (IGNORED_IDENTITY_PLATFORMS.has(platform)) { + this.log.warn({ key }, 'Ignored platform in identities, skipping') + return [] + } return (Array.isArray(values) ? values : [values]) .filter((v) => v && typeof v === 'string') diff --git a/services/libs/connectors/src/connectors/octolens/mapper.test.ts b/services/libs/connectors/src/connectors/octolens/mapper.test.ts index 5ec97408aa..74624f14c2 100644 --- a/services/libs/connectors/src/connectors/octolens/mapper.test.ts +++ b/services/libs/connectors/src/connectors/octolens/mapper.test.ts @@ -2,13 +2,18 @@ import { describe, expect, it } from 'vitest' import { MemberIdentityType, PlatformType } from '@crowd/types' -import { mapMentionToActivity, octolensActivitySchema, toIsoTimestamp } from './mapper' +import { + mapMentionToActivity, + octolensActivitySchema, + profileLinkIdentity, + toIsoTimestamp, +} from './mapper' import type { OctolensMention } from './mentionsClient' const BASE_MENTION: OctolensMention = { - id: 1, - sourceId: 'reddit_t3_1abc234', - url: 'https://example.com/post', + id: 200162933, + sourceId: 'IC_kwDOA1b2c3', + url: 'https://github.com/CrowdDotDev/crowd.dev/issues/12#issuecomment-1', title: 'Something about crowd.dev', body: 'post body', source: 'github', @@ -30,19 +35,20 @@ const BASE_MENTION: OctolensMention = { } describe('mapMentionToActivity', () => { - it('maps a supported source to the matching platform and a verified member identity', () => { + it('maps a github mention to a verified identity taken from the profile link', () => { const mapped = mapMentionToActivity(BASE_MENTION) expect(mapped?.platform).toBe(PlatformType.GITHUB) - expect(mapped?.keywordId).toBe(42) expect(mapped?.activity).toMatchObject({ type: 'mention', platform: PlatformType.GITHUB, - sourceId: 'reddit_t3_1abc234', - score: 6, + sourceId: '200162933', + score: 0, + channel: 'github', title: 'Something about crowd.dev', body: 'post body', - url: 'https://example.com/post', + url: 'https://github.com/CrowdDotDev/crowd.dev/issues/12#issuecomment-1', + timestamp: '2026-10-12T08:30:00.000Z', member: { displayName: 'The Octocat', identities: [ @@ -61,78 +67,156 @@ describe('mapMentionToActivity', () => { }) }) - it('matches source case-insensitively (historical rows can be uppercase)', () => { - const mapped = mapMentionToActivity({ ...BASE_MENTION, source: 'GITHUB' }) - - expect(mapped?.platform).toBe(PlatformType.GITHUB) - }) - - it.each(['twitter', 'linkedin'])('maps %s to its platform too', (source) => { - const mapped = mapMentionToActivity({ ...BASE_MENTION, source }) - - expect(mapped?.platform).toBe(source) - }) - - it('skips an unsupported source', () => { - expect(mapMentionToActivity({ ...BASE_MENTION, source: 'reddit' })).toBeNull() - }) - - it('skips a mention with no author identity', () => { - expect(mapMentionToActivity({ ...BASE_MENTION, author: null })).toBeNull() - }) - - it('falls back to the author handle when authorName is missing', () => { - const mapped = mapMentionToActivity({ ...BASE_MENTION, authorName: null }) - - expect(mapped?.activity.member.displayName).toBe('octocat') - }) - - it('omits attributes when authorUrl and authorAvatar are both missing', () => { + it('takes the linkedin handle from the profile link, not the display name in author', () => { const mapped = mapMentionToActivity({ ...BASE_MENTION, - authorUrl: null, - authorAvatar: null, + source: 'linkedin', + author: 'Daniel Kornaś', + authorName: 'Daniel Kornaś', + authorUrl: 'https://www.linkedin.com/in/daniel-kornas', }) - expect(mapped?.activity.member.attributes).toBeUndefined() + expect(mapped?.platform).toBe(PlatformType.LINKEDIN) + expect(mapped?.activity.member.identities[0].value).toBe('daniel-kornas') + expect(mapped?.activity.member.displayName).toBe('Daniel Kornaś') + expect(mapped?.activity.channel).toBe('linkedin') }) - it('converts the UTC-no-suffix timestamp to a proper ISO string', () => { - const mapped = mapMentionToActivity(BASE_MENTION) - - expect(mapped?.activity.timestamp).toBe('2026-10-12T08:30:00.000Z') + it.each(['https://twitter.com/DanKornas', 'https://x.com/DanKornas'])( + 'takes the twitter handle from %s', + (authorUrl) => { + const mapped = mapMentionToActivity({ ...BASE_MENTION, source: 'twitter', authorUrl }) + + expect(mapped?.platform).toBe(PlatformType.TWITTER) + expect(mapped?.activity.member.identities[0].value).toBe('DanKornas') + }, + ) + + it.each([ + ['a github app (bot)', 'https://github.com/apps/github-actions'], + ['a linkedin company page', 'https://www.linkedin.com/company/crowd-dev'], + ['a profile link with no handle', 'https://github.com/'], + ])('skips %s', (_name, authorUrl) => { + expect(mapMentionToActivity({ ...BASE_MENTION, authorUrl })).toBeNull() }) - it('passes through an absent keywordId as undefined rather than assuming it is set', () => { - const { keywordId, ...withoutKeywordId } = BASE_MENTION - void keywordId + it.each([ + ['reddit', 'https://github.com/octocat'], + ['podcasts', 'https://twitter.com/someshow'], + ['github', 'https://x.com/octocat'], + ])( + 'skips a %s mention whose profile link points at another platform (%s)', + (source, authorUrl) => { + expect(mapMentionToActivity({ ...BASE_MENTION, source, authorUrl })).toBeNull() + }, + ) + + it.each([undefined, null, ''])('skips a mention without an Octolens id (%s)', (id) => { + expect( + mapMentionToActivity({ ...BASE_MENTION, id } as unknown as typeof BASE_MENTION), + ).toBeNull() + }) - const mapped = mapMentionToActivity(withoutKeywordId as OctolensMention) + it.each(['github', 'twitter', 'linkedin'])( + 'skips a %s mention without a profile link instead of trusting author', + (source) => { + expect(mapMentionToActivity({ ...BASE_MENTION, source, authorUrl: null })).toBeNull() + }, + ) + + it.each([ + ['reddit', 'https://www.reddit.com/user/SuperUranus', PlatformType.REDDIT, 'SuperUranus'], + ['reddit_comment', 'https://old.reddit.com/u/someone', PlatformType.REDDIT, 'someone'], + ['dev', 'https://dev.to/nughes', PlatformType.DEVTO, 'nughes'], + [ + 'hackernews', + 'https://news.ycombinator.com/user?id=AnthonyMouse', + PlatformType.HACKERNEWS, + 'AnthonyMouse', + ], + [ + 'bluesky', + 'https://bsky.app/profile/cerestial.bsky.social', + PlatformType.BLUESKY, + 'cerestial.bsky.social', + ], + [ + 'tiktok', + 'https://www.tiktok.com/@indahpermata04632', + PlatformType.TIKTOK, + 'indahpermata04632', + ], + [ + 'instagram', + 'https://www.instagram.com/techfairyanna/', + PlatformType.INSTAGRAM, + 'techfairyanna', + ], + [ + 'youtube', + 'https://www.youtube.com/channel/UC1axavgir413w00rwBHPAtQ', + PlatformType.YOUTUBE, + 'UC1axavgir413w00rwBHPAtQ', + ], + ])( + 'emits a %s mention with the handle from its profile link', + (source, authorUrl, platform, handle) => { + const mapped = mapMentionToActivity({ + ...BASE_MENTION, + source, + authorUrl, + author: 'Display Name', + }) + + expect(mapped?.platform).toBe(platform) + expect(mapped?.activity.channel).toBe(source) + expect(mapped?.activity.member.identities[0]).toEqual({ + platform, + value: handle, + type: MemberIdentityType.USERNAME, + verified: true, + }) + }, + ) + + it.each([ + ['reddit', 'https://www.reddit.com/user/[deleted]'], + ['reddit', 'https://www.reddit.com/r/kubernetes'], + ['dev', 'https://dev.to/t/kubernetes'], + ['hackernews', 'https://news.ycombinator.com/item?id=50101892'], + ['bluesky', 'https://bsky.app/search?q=x'], + ['tiktok', 'https://www.tiktok.com/tag/kubernetes'], + ['instagram', 'https://www.instagram.com/p/DefkRhFsgpL/'], + ['youtube', 'https://www.youtube.com/@somehandle'], + ['instagram', 'https://www.tiktok.com/@someone'], + ])('skips a %s mention whose link %s is not a profile on that platform', (source, authorUrl) => { + expect(mapMentionToActivity({ ...BASE_MENTION, source, authorUrl })).toBeNull() + }) - expect(mapped?.keywordId).toBeUndefined() + it.each([ + ['stackoverflow', 'https://stackoverflow.com/users/1/someone'], + ['podcasts', 'https://www.spreaker.com/show/alacarta'], + ['news', null], + ['firehose', null], + ])('does not emit %s mentions', (source, authorUrl) => { + expect(mapMentionToActivity({ ...BASE_MENTION, source, authorUrl })).toBeNull() }) - it('captures every documented Mention field without a dedicated column in activity.attributes', () => { - const review = { - rating: 4, - ratingMax: 5, - verified: true, - response: null, - responseAt: null, - targetName: 'crowd.dev', - } + it('keeps the octolens fields and the mention context in activity.attributes', () => { const mapped = mapMentionToActivity({ ...BASE_MENTION, relevanceScore: 0, feedbackRelevant: 1, - imageUrl: 'https://example.com/preview.png', engagementMetrics: { likes: 10, replies: 2 }, - engagementObservedAt: '2026-10-12 09:00:00.000', - review, }) expect(mapped?.activity.attributes).toEqual({ - octolensMentionId: 1, + octolensMentionId: 200162933, + octolensSource: 'github', + postId: 'IC_kwDOA1b2c3', + keyword: 'crowd.dev', + keywordId: 42, + repoUrl: 'https://github.com/CrowdDotDev/crowd.dev', authorFollowers: 1200, relevance: 'relevant', relevanceScore: 0, @@ -141,61 +225,96 @@ describe('mapMentionToActivity', () => { language: 'english', tags: ['user_feedback'], keywords: [{ id: 42, keyword: 'crowd.dev', keywordTag: 'own_brand' }], - keywordId: 42, engaged: false, isReply: true, feedbackRelevant: 1, - imageUrl: 'https://example.com/preview.png', engagementMetrics: { likes: 10, replies: 2 }, - engagementObservedAt: '2026-10-12 09:00:00.000', - review: { rating: 4, ratingMax: 5, verified: true, targetName: 'crowd.dev' }, }) }) - it('omits null and absent optional fields instead of writing them as null', () => { + it('records the keyword chosen by keywordId when the mention lists several', () => { + const mapped = mapMentionToActivity({ + ...BASE_MENTION, + keywords: [ + { id: 7, keyword: 'other' }, + { id: 42, keyword: 'crowd.dev' }, + ], + keywordId: 42, + }) + + expect(mapped?.activity.attributes).toMatchObject({ keyword: 'crowd.dev', keywordId: 42 }) + }) + + it.each([ + ['https://github.com/octocat', undefined], + ['https://github.com/orgs/CrowdDotDev/discussions/5', undefined], + ['https://example.com/post', undefined], + ['https://www.github.com/owner/repo.git', 'https://github.com/owner/repo'], + ])('sets repoUrl for github url %s only when it names a repo', (url, repoUrl) => { + const mapped = mapMentionToActivity({ ...BASE_MENTION, url }) + + expect(mapped?.activity.attributes?.repoUrl).toBe(repoUrl) + }) + + it('omits null optional fields instead of writing them as null', () => { const mapped = mapMentionToActivity({ ...BASE_MENTION, authorFollowers: null, relevanceComment: null, sentiment: null, language: null, - relevanceScore: null, }) const attributes = mapped?.activity.attributes ?? {} - for (const key of [ - 'authorFollowers', - 'relevanceComment', - 'sentiment', - 'language', - 'relevanceScore', - 'review', - 'engagementMetrics', - 'imageUrl', - ]) { + for (const key of ['authorFollowers', 'relevanceComment', 'sentiment', 'language', 'review']) { expect(attributes).not.toHaveProperty(key) } }) - it('keeps false booleans and empty arrays in activity.attributes', () => { - const mapped = mapMentionToActivity({ ...BASE_MENTION, engaged: false, tags: [] }) - - expect(mapped?.activity.attributes).toMatchObject({ engaged: false, tags: [] }) + it('falls back to author, then the handle, for the display name', () => { + expect( + mapMentionToActivity({ ...BASE_MENTION, authorName: null })?.activity.member.displayName, + ).toBe('octocat') + expect( + mapMentionToActivity({ ...BASE_MENTION, authorName: '', author: '' })?.activity.member + .displayName, + ).toBe('octocat') }) - it('keeps authorFollowers on the activity only, never on member.reach', () => { - const mapped = mapMentionToActivity({ ...BASE_MENTION, source: 'linkedin' }) - - expect(mapped?.activity.attributes).toMatchObject({ authorFollowers: 1200 }) - expect(mapped?.activity.member).not.toHaveProperty('reach') - }) - - it('keeps the new attributes through octolensActivitySchema instead of stripping them', () => { + it('passes through octolensActivitySchema unchanged', () => { const mapped = mapMentionToActivity(BASE_MENTION) - const parsed = octolensActivitySchema.parse(mapped?.activity) + expect(octolensActivitySchema.parse(mapped?.activity)).toEqual(mapped?.activity) + }) +}) - expect(parsed.attributes).toMatchObject({ octolensMentionId: 1, sentiment: 'Positive' }) +describe('profileLinkIdentity', () => { + it.each([ + ['https://github.com/octocat', { platform: PlatformType.GITHUB, value: 'octocat' }], + ['https://x.com/someone', { platform: PlatformType.TWITTER, value: 'someone' }], + ['https://uk.linkedin.com/in/someone', { platform: PlatformType.LINKEDIN, value: 'someone' }], + ['https://github.com/apps/dependabot', null], + ['https://www.linkedin.com/in/alice%2Fbob', null], + ['https://www.linkedin.com/in/alice%5Cbob', null], + ['https://mobile.twitter.com/someone', { platform: PlatformType.TWITTER, value: 'someone' }], + ['https://x.com/i/user/123', null], + ['https://x.com/compose/post', null], + ['https://twitter.com/login', null], + ['https://x.com/signup', null], + ['https://github.com/user-', null], + ['https://github.com/user--name', null], + ['https://github.com/user-name', { platform: PlatformType.GITHUB, value: 'user-name' }], + ['https://twitter.com/intent/user?screen_name=foo', null], + ['https://twitter.com/hashtag/kubernetes', null], + [ + 'https://www.linkedin.com/in/j%C3%B6rg-x/', + { platform: PlatformType.LINKEDIN, value: 'jörg-x' }, + ], + ['https://www.reddit.com/user/someone', { platform: PlatformType.REDDIT, value: 'someone' }], + ['https://example.com/someone', undefined], + [null, undefined], + ])('reads %s', (link, expected) => { + expect(profileLinkIdentity(link)).toEqual(expected) }) }) diff --git a/services/libs/connectors/src/connectors/octolens/mapper.ts b/services/libs/connectors/src/connectors/octolens/mapper.ts index 5cb164a95a..aaddcc8b3d 100644 --- a/services/libs/connectors/src/connectors/octolens/mapper.ts +++ b/services/libs/connectors/src/connectors/octolens/mapper.ts @@ -4,17 +4,80 @@ import { MemberIdentityType, PlatformType } from '@crowd/types' import type { OctolensMention } from './mentionsClient' -// Only these 3 of Octolens's ~13 source values are supported (D8); the rest are skipped -// and logged in the mapper below, not thrown. +// Only sources whose author is a person: podcasts, news, firehose, newsletter and review sites +// name a show or publication. const SOURCE_TO_PLATFORM: Record = { github: PlatformType.GITHUB, twitter: PlatformType.TWITTER, linkedin: PlatformType.LINKEDIN, + reddit: PlatformType.REDDIT, + reddit_comment: PlatformType.REDDIT, + dev: PlatformType.DEVTO, + hackernews: PlatformType.HACKERNEWS, + stackoverflow: PlatformType.STACKOVERFLOW, + medium: PlatformType.MEDIUM, + producthunt: PlatformType.PRODUCTHUNT, + facebook: PlatformType.FACEBOOK, + youtube: PlatformType.YOUTUBE, + bluesky: PlatformType.BLUESKY, + tiktok: PlatformType.TIKTOK, + instagram: PlatformType.INSTAGRAM, } -// Matches the existing MENTION score already set for github/twitter/linkedin's own -// activity-scoring grids (all 6) — see services/libs/integrations/*/grid.ts. -const OCTOLENS_MENTION_SCORE = 6 +// Every source whose live mentions carry a verifiable profile link; stackoverflow, medium, +// producthunt and facebook stay mapped but off until Octolens actually sends them. +const ENABLED_PLATFORMS = new Set([ + PlatformType.GITHUB, + PlatformType.TWITTER, + PlatformType.LINKEDIN, + PlatformType.REDDIT, + PlatformType.DEVTO, + PlatformType.HACKERNEWS, + PlatformType.BLUESKY, + PlatformType.TIKTOK, + PlatformType.INSTAGRAM, + PlatformType.YOUTUBE, +]) + +const GITHUB_NON_OWNER_PATHS = new Set([ + 'about', + 'account', + 'advisories', + 'apps', + 'codespaces', + 'collections', + 'copilot', + 'customer-stories', + 'dashboard', + 'enterprise', + 'events', + 'explore', + 'features', + 'issues', + 'login', + 'marketplace', + 'new', + 'notifications', + 'organizations', + 'orgs', + 'pricing', + 'pulls', + 'readme', + 'resources', + 'search', + 'security', + 'settings', + 'site', + 'solutions', + 'sponsors', + 'stars', + 'topics', + 'trending', + 'users', +]) + +// Mentions are for identity enrichment only, so they never move a member's engagement score. +const OCTOLENS_MENTION_SCORE = 0 const identitySchema = z.object({ platform: z.string(), @@ -38,6 +101,7 @@ export const octolensActivitySchema = z.object({ title: z.string().optional(), body: z.string().optional(), url: z.string().optional(), + channel: z.string().optional(), attributes: z.record(z.unknown()).optional(), member: memberSchema, }) @@ -46,15 +110,15 @@ export type OctolensActivity = z.infer export interface MappedOctolensMention { platform: PlatformType - keywordId: number | undefined activity: OctolensActivity } export function mapMentionToActivity(mention: OctolensMention): MappedOctolensMention | null { - const platform = SOURCE_TO_PLATFORM[mention.source.toLowerCase()] - if (!platform || !mention.author) { + const author = authorIdentity(mention) + if (!author || !ENABLED_PLATFORMS.has(author.platform) || !hasMentionId(mention.id)) { return null } + const { platform, value: identityValue } = author // Each attribute name is a top-level key, wrapped per-platform underneath — matches the // GitHub connector's own memberAttributesSchema shape, not { [platform]: { url, ... } }. @@ -66,6 +130,11 @@ export function mapMentionToActivity(mention: OctolensMention): MappedOctolensMe // reach.linkedin from connections, and these per-mention snapshots can arrive out of order. const activityAttributes = withoutNullish({ octolensMentionId: mention.id, + octolensSource: mention.source, + postId: mention.sourceId, + keyword: matchedKeyword(mention), + repoUrl: platform === PlatformType.GITHUB ? parseGithubRepoUrl(mention.url) : undefined, + subreddit: platform === PlatformType.REDDIT ? parseSubreddit(mention.url) : undefined, authorFollowers: mention.authorFollowers, relevance: mention.relevance, relevanceScore: mention.relevanceScore, @@ -86,23 +155,23 @@ export function mapMentionToActivity(mention: OctolensMention): MappedOctolensMe return { platform, - keywordId: mention.keywordId, activity: { type: 'mention', platform, timestamp: toIsoTimestamp(mention.timestamp), - sourceId: mention.sourceId, + sourceId: String(mention.id), score: OCTOLENS_MENTION_SCORE, title: mention.title ?? undefined, body: mention.body ?? undefined, url: mention.url, + channel: mention.source, attributes: Object.keys(activityAttributes).length > 0 ? activityAttributes : undefined, member: { - displayName: mention.authorName ?? mention.author, + displayName: mention.authorName || mention.author || identityValue, identities: [ { platform, - value: mention.author, + value: identityValue, type: MemberIdentityType.USERNAME, verified: true, }, @@ -113,6 +182,218 @@ export function mapMentionToActivity(mention: OctolensMention): MappedOctolensMe } } +interface AuthorIdentity { + platform: PlatformType + value: string +} + +// GitHub, X/Twitter and LinkedIn identities come from the author's profile link: Octolens's +// `author` is a display name on LinkedIn, and a github.com/apps/ link is a bot, not a person. +function authorIdentity(mention: OctolensMention): AuthorIdentity | null { + const platform = SOURCE_TO_PLATFORM[mention.source.toLowerCase()] + if (!platform) return null + + if (PROFILE_LINK_PLATFORMS.has(platform)) { + const fromLink = profileLinkIdentity(mention.authorUrl) + return fromLink && fromLink.platform === platform ? fromLink : null + } + + return mention.author ? { platform, value: mention.author } : null +} + +const PROFILE_LINK_PLATFORMS = new Set([ + PlatformType.GITHUB, + PlatformType.TWITTER, + PlatformType.LINKEDIN, + PlatformType.REDDIT, + PlatformType.DEVTO, + PlatformType.HACKERNEWS, + PlatformType.BLUESKY, + PlatformType.TIKTOK, + PlatformType.INSTAGRAM, + PlatformType.YOUTUBE, +]) + +const GITHUB_LOGIN_RE = /^(?=.{1,39}$)[A-Za-z0-9]+(?:-[A-Za-z0-9]+)*$/ +const TWITTER_HANDLE_RE = /^[A-Za-z0-9_]{1,15}$/ +const TWITTER_NON_HANDLE_PATHS = new Set([ + 'compose', + 'explore', + 'hashtag', + 'home', + 'i', + 'intent', + 'login', + 'logout', + 'messages', + 'notifications', + 'search', + 'settings', + 'share', + 'signup', + 'tos', +]) + +// Reddit's `[deleted]` fails REDDIT_USERNAME_RE, so deleted accounts never share one identity. +const REDDIT_USERNAME_RE = /^[A-Za-z0-9_-]{3,20}$/ +const DEVTO_USERNAME_RE = /^[A-Za-z0-9_]{1,30}$/ +const DEVTO_NON_USER_PATHS = new Set([ + 'about', + 'code-of-conduct', + 'enter', + 'faq', + 'latest', + 'privacy', + 'search', + 'settings', + 't', + 'terms', + 'top', +]) +const HACKERNEWS_USERNAME_RE = /^[A-Za-z0-9_-]{2,15}$/ +const BLUESKY_HANDLE_RE = /^(?:did:[a-z]+:[A-Za-z0-9._:-]+|(?:[A-Za-z0-9-]+\.)+[A-Za-z]{2,})$/ +const TIKTOK_HANDLE_RE = /^@([A-Za-z0-9_.]{2,24})$/ +const INSTAGRAM_USERNAME_RE = /^[A-Za-z0-9_.]{1,30}$/ +const INSTAGRAM_NON_USER_PATHS = new Set([ + 'accounts', + 'direct', + 'explore', + 'p', + 'reel', + 'reels', + 'stories', + 'tv', +]) +const YOUTUBE_CHANNEL_ID_RE = /^UC[A-Za-z0-9_-]{22}$/ + +// undefined: not a known profile link; null: a known host, but no usable person. +export function profileLinkIdentity( + raw: string | null | undefined, +): AuthorIdentity | null | undefined { + const url = parseUrl(raw) + if (!url) return undefined + const host = url.hostname.toLowerCase().replace(/^(www|mobile|m|old|new)\./, '') + const segments = pathSegments(url).map(decodeSegment) + + if (host === 'github.com') { + const [login] = segments + return login && GITHUB_LOGIN_RE.test(login) && !GITHUB_NON_OWNER_PATHS.has(login.toLowerCase()) + ? { platform: PlatformType.GITHUB, value: login } + : null + } + if (host === 'twitter.com' || host === 'x.com') { + const [handle] = segments + return handle && + TWITTER_HANDLE_RE.test(handle) && + !TWITTER_NON_HANDLE_PATHS.has(handle.toLowerCase()) + ? { platform: PlatformType.TWITTER, value: handle } + : null + } + if (host === 'linkedin.com' || host.endsWith('.linkedin.com')) { + const [kind, handle] = segments + return kind === 'in' && handle && !/[\\/]/.test(handle) + ? { platform: PlatformType.LINKEDIN, value: handle } + : null + } + if (host === 'reddit.com' || host.endsWith('.reddit.com')) { + const [kind, username] = segments + return (kind === 'user' || kind === 'u') && username && REDDIT_USERNAME_RE.test(username) + ? { platform: PlatformType.REDDIT, value: username } + : null + } + if (host === 'dev.to') { + const [username] = segments + return username && + DEVTO_USERNAME_RE.test(username) && + !DEVTO_NON_USER_PATHS.has(username.toLowerCase()) + ? { platform: PlatformType.DEVTO, value: username } + : null + } + if (host === 'news.ycombinator.com') { + const username = url.searchParams.get('id') + return segments[0] === 'user' && username && HACKERNEWS_USERNAME_RE.test(username) + ? { platform: PlatformType.HACKERNEWS, value: username } + : null + } + if (host === 'bsky.app') { + const [kind, handle] = segments + return kind === 'profile' && handle && BLUESKY_HANDLE_RE.test(handle) + ? { platform: PlatformType.BLUESKY, value: handle } + : null + } + if (host === 'tiktok.com') { + const match = segments[0]?.match(TIKTOK_HANDLE_RE) + return match ? { platform: PlatformType.TIKTOK, value: match[1] } : null + } + if (host === 'instagram.com') { + const [username] = segments + return username && + INSTAGRAM_USERNAME_RE.test(username) && + !INSTAGRAM_NON_USER_PATHS.has(username.toLowerCase()) + ? { platform: PlatformType.INSTAGRAM, value: username } + : null + } + if (host === 'youtube.com') { + const [kind, channelId] = segments + return kind === 'channel' && channelId && YOUTUBE_CHANNEL_ID_RE.test(channelId) + ? { platform: PlatformType.YOUTUBE, value: channelId } + : null + } + return undefined +} + +function decodeSegment(segment: string): string { + try { + return decodeURIComponent(segment) + } catch { + return segment + } +} + +function hasMentionId(id: unknown): boolean { + return Number.isSafeInteger(id) || (typeof id === 'string' && id.length > 0) +} + +function matchedKeyword(mention: OctolensMention): string | undefined { + const keywords = Array.isArray(mention.keywords) ? mention.keywords : [] + const match = keywords.find((k) => k?.id === mention.keywordId) + const text = typeof match?.keyword === 'string' ? match.keyword.trim() : '' + return text || undefined +} + +function parseUrl(raw: string | null | undefined): URL | null { + if (!raw) return null + try { + return new URL(raw) + } catch { + return null + } +} + +function pathSegments(url: URL): string[] { + return url.pathname.split('/').filter((segment) => segment.length > 0) +} + +export function parseGithubRepoUrl(raw: string | null | undefined): string | undefined { + const url = parseUrl(raw) + if (!url || !['github.com', 'www.github.com'].includes(url.hostname.toLowerCase())) { + return undefined + } + const [owner, rawRepo] = pathSegments(url) + const repo = rawRepo?.replace(/\.git$/i, '') + if (!owner || !repo || GITHUB_NON_OWNER_PATHS.has(owner.toLowerCase())) { + return undefined + } + return `https://github.com/${owner}/${repo}` +} + +export function parseSubreddit(raw: string | null | undefined): string | undefined { + const url = parseUrl(raw) + if (!url || !/(^|\.)reddit\.com$/i.test(url.hostname)) return undefined + const [first, subreddit] = pathSegments(url) + return first?.toLowerCase() === 'r' && subreddit ? subreddit : undefined +} + function withoutNullish(obj: Record): Record { return Object.fromEntries(Object.entries(obj).filter(([, v]) => v !== null && v !== undefined)) } diff --git a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts index f0e092a7d9..64cfc9654e 100644 --- a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts +++ b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.test.ts @@ -1,4 +1,4 @@ -import { beforeEach, describe, expect, it, vi } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { Logger } from '@crowd/logging' import { PlatformType } from '@crowd/types' @@ -24,9 +24,11 @@ const OCTOLENS_CREDENTIAL: Credential = { data: { token: 'test-token' }, } -const LOG = { info: vi.fn(), warn: vi.fn(), error: vi.fn() } as unknown as Logger +const LOG = { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() } as unknown as Logger +// Like the live API: the keyword lives only in keywords[], with no top-level keywordId. function mention(overrides: Partial = {}): OctolensMention { + const { keywordId = 1, ...rest } = overrides return { id: 1, sourceId: 'src-1', @@ -37,7 +39,7 @@ function mention(overrides: Partial = {}): OctolensMention { timestamp: '2026-10-12 08:00:00.000', author: 'octocat', authorName: 'The Octocat', - authorUrl: null, + authorUrl: 'https://github.com/octocat', authorAvatar: null, authorFollowers: null, relevance: 'relevant', @@ -45,11 +47,10 @@ function mention(overrides: Partial = {}): OctolensMention { sentiment: null, language: null, tags: [], - keywords: [], + keywords: keywordId === undefined ? [] : [{ id: keywordId, keyword: `kw-${keywordId}` }], engaged: false, isReply: false, - keywordId: 1, - ...overrides, + ...rest, } } @@ -104,8 +105,17 @@ function pollFields({ startDate, cursor, pendingMaxTimestamp }: Record { mockedFetchMentionsPage.mockReset() + vi.useFakeTimers({ now: new Date(NOW), toFake: ['Date'] }) +}) + +afterEach(() => { + vi.useRealTimers() }) describe('runMentionPoll', () => { @@ -126,6 +136,7 @@ describe('runMentionPoll', () => { ]), ) expect(watermarks.map(pollFields)).toEqual([ + INITIAL_CUTOFF, { startDate: '2026-10-12T08:00:00.000Z', cursor: null, pendingMaxTimestamp: null }, ]) }) @@ -155,14 +166,19 @@ describe('runMentionPoll', () => { expect(outcome).toEqual({ complete: true }) expect(emitted).toHaveLength(1) - expect((emitted[0].records[0] as { sourceId: string }).sourceId).toBe('ok') + expect((emitted[0].records[0] as { attributes: { postId: string } }).attributes.postId).toBe( + 'ok', + ) expect(watermarks.map(pollFields)).toEqual([ + INITIAL_CUTOFF, { startDate: '2026-10-12T08:00:00.000Z', cursor: null, pendingMaxTimestamp: null }, ]) }) it('skips an unsupported source without throwing or emitting it', async () => { - mockedFetchMentionsPage.mockResolvedValue(page([mention({ source: 'reddit' })])) + mockedFetchMentionsPage.mockResolvedValue( + page([mention({ source: 'podcasts', authorUrl: 'https://www.spreaker.com/show/x' })]), + ) const { ctx, emitted } = makeCtx() const outcome = await runMentionPoll(ctx) @@ -180,10 +196,8 @@ describe('runMentionPoll', () => { expect(emitted).toHaveLength(0) }) - it('skips a mention with no keywordId without calling resolveKeywordTarget', async () => { - const { keywordId, ...withoutKeywordId } = mention() - void keywordId - mockedFetchMentionsPage.mockResolvedValue(page([withoutKeywordId as OctolensMention])) + it('skips a mention with no keywords without calling resolveKeywordTarget', async () => { + mockedFetchMentionsPage.mockResolvedValue(page([mention({ keywords: [] })])) const resolveKeywordTarget = vi.fn(async () => target('segment-1')) const { ctx, emitted } = makeCtx({ resolveKeywordTarget }) @@ -196,7 +210,11 @@ describe('runMentionPoll', () => { it('skips a mention that fails to map (e.g. a malformed field) instead of crashing the page', async () => { mockedFetchMentionsPage.mockResolvedValue( page([ - { ...mention({ sourceId: 'bad' }), source: undefined } as unknown as OctolensMention, + { + ...mention({ sourceId: 'bad' }), + source: undefined, + authorUrl: undefined, + } as unknown as OctolensMention, mention({ sourceId: 'ok' }), ]), ) @@ -206,7 +224,9 @@ describe('runMentionPoll', () => { expect(outcome).toEqual({ complete: true }) expect(emitted).toHaveLength(1) - expect((emitted[0].records[0] as { sourceId: string }).sourceId).toBe('ok') + expect((emitted[0].records[0] as { attributes: { postId: string } }).attributes.postId).toBe( + 'ok', + ) }) it('still advances the watermark past a mention whose timestamp parsed but mapping failed', async () => { @@ -215,14 +235,17 @@ describe('runMentionPoll', () => { { ...mention({ timestamp: '2026-10-12 09:00:00.000' }), source: undefined, + authorUrl: undefined, } as unknown as OctolensMention, ]), ) - const { ctx, watermarks } = makeCtx() + const { ctx, emitted, watermarks } = makeCtx() await runMentionPoll(ctx) + expect(emitted).toHaveLength(0) expect(watermarks.map(pollFields)).toEqual([ + INITIAL_CUTOFF, { startDate: '2026-10-12T09:00:00.000Z', cursor: null, pendingMaxTimestamp: null }, ]) }) @@ -240,7 +263,9 @@ describe('runMentionPoll', () => { expect(outcome).toEqual({ complete: true }) expect(emitted).toHaveLength(1) - expect((emitted[0].records[0] as { sourceId: string }).sourceId).toBe('ok') + expect((emitted[0].records[0] as { attributes: { postId: string } }).attributes.postId).toBe( + 'ok', + ) }) it('skips a null page entry instead of crashing the whole page', async () => { @@ -253,7 +278,9 @@ describe('runMentionPoll', () => { expect(outcome).toEqual({ complete: true }) expect(emitted).toHaveLength(1) - expect((emitted[0].records[0] as { sourceId: string }).sourceId).toBe('ok') + expect((emitted[0].records[0] as { attributes: { postId: string } }).attributes.postId).toBe( + 'ok', + ) }) it('propagates a resolveKeywordTarget failure instead of swallowing it as a skip', async () => { @@ -265,17 +292,21 @@ describe('runMentionPoll', () => { }) await expect(runMentionPoll(ctx)).rejects.toThrow('db connection lost') - // No watermark commit either — a transient failure must not advance past this mention. - expect(watermarks).toHaveLength(0) + // Only the first run's cutoff: a transient failure must not advance past this mention. + expect(watermarks.map(pollFields)).toEqual([INITIAL_CUTOFF]) }) it('still advances the watermark past a skipped mention', async () => { - mockedFetchMentionsPage.mockResolvedValue(page([mention({ source: 'reddit' })])) - const { ctx, watermarks } = makeCtx() + mockedFetchMentionsPage.mockResolvedValue( + page([mention({ source: 'podcasts', authorUrl: 'https://www.spreaker.com/show/x' })]), + ) + const { ctx, emitted, watermarks } = makeCtx() await runMentionPoll(ctx) + expect(emitted).toHaveLength(0) expect(watermarks.map(pollFields)).toEqual([ + INITIAL_CUTOFF, { startDate: '2026-10-12T08:00:00.000Z', cursor: null, pendingMaxTimestamp: null }, ]) }) @@ -398,32 +429,104 @@ describe('runMentionPoll', () => { }) }) -describe('runMentionPoll keyword backfill', () => { - const MAPPED = async (keywordId: number) => (keywordId === 7 ? target('segment-7') : null) +describe('runMentionPoll first run', () => { + it('saves its start date before fetching, so a failed first run retries the same window', async () => { + mockedFetchMentionsPage.mockRejectedValueOnce(new Error('octolens unavailable')) + const first = makeCtx() + + await expect(runMentionPoll(first.ctx)).rejects.toThrow('octolens unavailable') + expect(first.watermarks.map(pollFields)).toEqual([INITIAL_CUTOFF]) + + vi.setSystemTime(new Date('2026-10-02T00:00:00.000Z')) + mockedFetchMentionsPage.mockResolvedValue(page([])) + const retry = makeCtx({ watermark: first.watermarks[0] }) + + await runMentionPoll(retry.ctx) + expect(mockedFetchMentionsPage).toHaveBeenLastCalledWith('test-token', { + startDate: NOW, + cursor: undefined, + }) + }) + + it('keeps its first start date when it stops early, so the next run resumes the same window', async () => { + let calls = 0 + mockedFetchMentionsPage.mockResolvedValue(page([], 'cursor-2')) + const { ctx, watermarks } = makeCtx({ hasRunBudget: () => calls++ === 0 }) + + expect(await runMentionPoll(ctx)).toEqual({ complete: false }) + expect(watermarks[watermarks.length - 1]).toMatchObject({ startDate: NOW, cursor: 'cursor-2' }) + }) + + it('starts at now instead of fetching the whole history', async () => { + mockedFetchMentionsPage.mockResolvedValue(page([])) + const { ctx, watermarks } = makeCtx() + + expect(await runMentionPoll(ctx)).toEqual({ complete: true }) + expect(mockedFetchMentionsPage).toHaveBeenCalledWith('test-token', { + startDate: NOW, + cursor: undefined, + }) + expect(watermarks[watermarks.length - 1]).toMatchObject({ startDate: NOW, backfill: [] }) + }) +}) + +describe('runMentionPoll multi-keyword mentions', () => { + it('routes a mention by its first mapped keyword', async () => { + mockedFetchMentionsPage.mockResolvedValue( + page([ + mention({ + keywords: [ + { id: 5, keyword: 'unmapped' }, + { id: 7, keyword: 'second' }, + { id: 9, keyword: 'third' }, + ], + }), + ]), + ) + const { ctx, emitted } = makeCtx({ + resolveKeywordTarget: async (keywordId) => + keywordId === 5 ? null : target(`segment-${keywordId}`), + }) + + await runMentionPoll(ctx) + + expect(emitted).toHaveLength(1) + expect(emitted[0].segmentId).toBe('segment-7') + expect(emitted[0].records[0]).toMatchObject({ + attributes: { keywordId: 7, keyword: 'second' }, + }) + }) - it('records a keyword whose mentions were skipped for lack of a mapping', async () => { + it('never queues a backfill on its own when a keyword is unmapped', async () => { mockedFetchMentionsPage.mockResolvedValue(page([mention({ keywordId: 7 })])) const { ctx, emitted, watermarks } = makeCtx({ resolveKeywordTarget: async () => null }) - const outcome = await runMentionPoll(ctx) + await runMentionPoll(ctx) - expect(outcome).toEqual({ complete: true }) expect(emitted).toEqual([]) - expect(watermarks[watermarks.length - 1]).toMatchObject({ unmappedKeywords: [7], backfill: [] }) + expect(watermarks[watermarks.length - 1]).toEqual( + expect.not.objectContaining({ unmappedKeywords: expect.anything() }), + ) + expect(watermarks[watermarks.length - 1]).toMatchObject({ backfill: [] }) }) +}) + +describe('runMentionPoll keyword backfill (queued manually)', () => { + const MAPPED = async (keywordId: number) => (keywordId === 7 ? target('segment-7') : null) - it("backfills a newly mapped keyword's whole history, filtered to that keyword", async () => { + it("fetches a queued keyword's whole history, filtered to that keyword", async () => { mockedFetchMentionsPage .mockResolvedValueOnce(page([])) .mockResolvedValueOnce(page([mention({ keywordId: 7, sourceId: 'old-1' })])) const { ctx, emitted, watermarks } = makeCtx({ resolveKeywordTarget: MAPPED, - watermark: { startDate: '2026-10-12T08:00:00.000Z', unmappedKeywords: [7, 8] }, + watermark: { + startDate: '2026-10-12T08:00:00.000Z', + backfill: [{ keywordId: 7, cursor: null }], + }, }) - const outcome = await runMentionPoll(ctx) - - expect(outcome).toEqual({ complete: true }) + expect(await runMentionPoll(ctx)).toEqual({ complete: true }) expect(mockedFetchMentionsPage).toHaveBeenLastCalledWith('test-token', { keywordIds: [7], cursor: undefined, @@ -433,7 +536,6 @@ describe('runMentionPoll keyword backfill', () => { ]) expect(watermarks[watermarks.length - 1]).toMatchObject({ startDate: '2026-10-12T08:00:00.000Z', - unmappedKeywords: [8], backfill: [], }) }) @@ -479,7 +581,10 @@ describe('runMentionPoll keyword backfill', () => { mockedFetchMentionsPage.mockResolvedValue(page([], 'poll-cursor')) const { ctx, watermarks } = makeCtx({ resolveKeywordTarget: MAPPED, - watermark: { startDate: '2026-10-12T08:00:00.000Z', unmappedKeywords: [7] }, + watermark: { + startDate: '2026-10-12T08:00:00.000Z', + backfill: [{ keywordId: 7, cursor: null }], + }, hasRunBudget: () => calls++ === 0, }) @@ -487,12 +592,54 @@ describe('runMentionPoll keyword backfill', () => { expect(mockedFetchMentionsPage).toHaveBeenCalledTimes(1) expect(watermarks[watermarks.length - 1]).toMatchObject({ cursor: 'poll-cursor', - unmappedKeywords: [], backfill: [{ keywordId: 7, cursor: null }], }) }) - it('drops a backfill whose keyword lost its mapping, so it restarts when mapped again', async () => { + it('routes a backfilled mention like a polled one, by its first mapped keyword', async () => { + mockedFetchMentionsPage.mockResolvedValueOnce(page([])).mockResolvedValueOnce( + page([ + mention({ + keywords: [ + { id: 3, keyword: 'other' }, + { id: 7, keyword: 'backfilled' }, + ], + }), + ]), + ) + const { ctx, emitted } = makeCtx({ + resolveKeywordTarget: async (keywordId) => target(`segment-${keywordId}`), + watermark: { + startDate: '2026-10-12T08:00:00.000Z', + backfill: [{ keywordId: 7, cursor: null }], + }, + }) + + await runMentionPoll(ctx) + + expect(emitted).toEqual([expect.objectContaining({ segmentId: 'segment-3' })]) + }) + + it('skips a null entry on a backfill page instead of blocking the backfill', async () => { + mockedFetchMentionsPage + .mockResolvedValueOnce(page([])) + .mockResolvedValueOnce( + page([null as unknown as OctolensMention, mention({ keywordId: 7, sourceId: 'ok' })]), + ) + const { ctx, emitted, watermarks } = makeCtx({ + resolveKeywordTarget: MAPPED, + watermark: { + startDate: '2026-10-12T08:00:00.000Z', + backfill: [{ keywordId: 7, cursor: null }], + }, + }) + + expect(await runMentionPoll(ctx)).toEqual({ complete: true }) + expect(emitted).toHaveLength(1) + expect(watermarks[watermarks.length - 1]).toMatchObject({ backfill: [] }) + }) + + it('drops a queued backfill whose keyword has no mapping', async () => { mockedFetchMentionsPage.mockResolvedValue(page([])) const { ctx, watermarks } = makeCtx({ resolveKeywordTarget: async () => null, @@ -504,10 +651,7 @@ describe('runMentionPoll keyword backfill', () => { expect(await runMentionPoll(ctx)).toEqual({ complete: true }) expect(mockedFetchMentionsPage).toHaveBeenCalledTimes(1) - expect(watermarks[watermarks.length - 1]).toMatchObject({ - unmappedKeywords: [7], - backfill: [], - }) + expect(watermarks[watermarks.length - 1]).toMatchObject({ backfill: [] }) }) it('stops a backfill after a page cap so it stays within the shared request budget', async () => { @@ -567,13 +711,12 @@ describe('runMentionPoll keyword backfill', () => { ]) }) - it('ignores malformed keyword tracking in a stored watermark', async () => { + it('ignores malformed backfill entries in a stored watermark', async () => { mockedFetchMentionsPage.mockResolvedValue(page([])) const { ctx, watermarks } = makeCtx({ resolveKeywordTarget: async () => null, watermark: { startDate: '2026-10-12T08:00:00.000Z', - unmappedKeywords: ['7', null], backfill: [ { keywordId: 'x', cursor: null }, { keywordId: 9, cursor: 5 }, @@ -583,6 +726,6 @@ describe('runMentionPoll keyword backfill', () => { expect(await runMentionPoll(ctx)).toEqual({ complete: true }) expect(mockedFetchMentionsPage).toHaveBeenCalledTimes(1) - expect(watermarks[watermarks.length - 1]).toMatchObject({ unmappedKeywords: [], backfill: [] }) + expect(watermarks[watermarks.length - 1]).toMatchObject({ backfill: [] }) }) }) diff --git a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts index cc6208956b..c5b6e61831 100644 --- a/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts +++ b/services/libs/connectors/src/connectors/octolens/syncs/mentionPoll.ts @@ -21,9 +21,7 @@ interface OctolensWatermark { // Running max timestamp for the poll cycle in progress, separate from startDate (which // must stay frozen all cycle) so a resumed cycle doesn't lose earlier pages' progress. pendingMaxTimestamp: string | null - // Keywords whose mentions were skipped for lack of a mapping. startDate has already moved past - // those mentions, so once a keyword gets mapped its whole history is re-fetched via `backfill`. - unmappedKeywords: number[] + // Filled only by the manual backfill script; each keyword's whole history is re-fetched. backfill: KeywordBackfill[] } @@ -33,9 +31,6 @@ function readWatermark(raw: Record | null): OctolensWatermark { cursor: typeof raw?.cursor === 'string' ? raw.cursor : null, pendingMaxTimestamp: typeof raw?.pendingMaxTimestamp === 'string' ? raw.pendingMaxTimestamp : null, - unmappedKeywords: Array.isArray(raw?.unmappedKeywords) - ? raw.unmappedKeywords.filter((id): id is number => Number.isInteger(id)) - : [], backfill: Array.isArray(raw?.backfill) ? raw.backfill .filter( @@ -54,28 +49,19 @@ export async function runMentionPoll(ctx: SyncContext): Promise item.keywordId === keywordId)) { - backfill.push({ keywordId, cursor: null }) - } + if (await ctx.deps.resolveKeywordTarget(item.keywordId)) { + backfill.push(item) + } else { + ctx.log.warn({ keywordId: item.keywordId }, 'dropping octolens keyword backfill: no mapping') } } - // Known limitation: Octolens has no "discovered at" field, so a historical mention it adds - // behind our watermark for an already-mapped keyword is missed. - // Passed on every page this run — undocumented whether Octolens's cursor implicitly - // preserves the original filters, so don't rely on it. - const startDate = watermark.startDate ?? undefined + // Passed on every page this run: undocumented whether Octolens's cursor keeps the original filters. + const startDate = watermark.startDate ?? firstRunStartDate() let cursor: string | undefined = watermark.cursor ?? undefined - let maxTimestampSeen = watermark.pendingMaxTimestamp ?? watermark.startDate + let maxTimestampSeen = watermark.pendingMaxTimestamp ?? startDate // Not read via ctx.http (its token pool is unseeded here, see Manifest), so // sync_run_summary's requestCount always reads 0 for octolens — logged separately below. let pagesFetched = 0 @@ -83,7 +69,6 @@ export async function runMentionPoll(ctx: SyncContext): Promise) => ctx.commitWatermark({ ...poll, - unmappedKeywords: [...unmappedKeywords], backfill: backfill.map((item) => ({ ...item })), }) @@ -96,18 +81,11 @@ export async function runMentionPoll(ctx: SyncContext): Promise[] } >() for (const mention of mentions) { - let mapped: MappedOctolensMention | null - // A malformed field must not crash the whole page, so wrap mapping in try/catch — - // but update the watermark first so one bad mention type can't stall it forever. try { - onTimestamp?.(toIsoTimestamp(mention.timestamp)) - mapped = mapMentionToActivity(mention) - // Catches a numeric title/sourceId etc that passes the mapper but would otherwise - // crash createEmit's own schema parse, uncaught, outside this loop. - if (mapped) octolensActivitySchema.parse(mapped.activity) + const isoTimestamp = toIsoTimestamp(mention.timestamp) + onTimestamp?.(isoTimestamp) } catch (err) { - // Optional chaining: a null/non-object page.data entry must not also crash this - // handler by dereferencing it here. + // Optional chaining: a null/non-object page.data entry must not crash this handler too. ctx.log.warn( { sourceId: mention?.sourceId, errMsg: (err as Error).message }, 'skipping octolens mention: failed to process (malformed provider data)', @@ -115,36 +93,46 @@ export async function runMentionPoll(ctx: SyncContext): Promise): Promise): Promise): Promise k?.id) : []), + ] + return [...new Set(ids.filter((id): id is number => Number.isInteger(id)))] +} + +// Older mentions arrive only through a manually queued backfill. +function firstRunStartDate(): string { + return new Date().toISOString() +} diff --git a/services/libs/connectors/src/emit.test.ts b/services/libs/connectors/src/emit.test.ts index 17c2b15562..e76c6d5dce 100644 --- a/services/libs/connectors/src/emit.test.ts +++ b/services/libs/connectors/src/emit.test.ts @@ -313,4 +313,78 @@ describe('createEmit', () => { ]) expect(deps.publishResult).not.toHaveBeenCalled() }) + + describe('channel', () => { + const SCHEMA_WITH_CHANNEL = SCHEMA.extend({ channel: z.string().optional() }) + + function emitterWith(schema: z.ZodType>, emitEnabled = true) { + const deps = makeDeps() + const emitter = createEmit({ + publishResult: deps.publishResult, + sinkEmitter: deps.sinkEmitter, + recordShadow: deps.recordShadow, + unit: { ...UNIT, emitEnabled }, + segmentId: 'segment-default', + schema, + log: LOG, + }) + const publishedChannels = () => + deps.publishResult.mock.calls.map( + ([, result]) => (result as unknown as { data: { channel: unknown } }).data.channel, + ) + return { deps, emitter, publishedChannels } + } + + it('uses the sync unit channel for a record without one (GitHub-style schema)', async () => { + const { emitter, publishedChannels } = emitterWith(SCHEMA) + + await emitter.emit([record('a')]) + + expect(publishedChannels()).toEqual(['octolens-global']) + }) + + it('overwrites a stray channel when the schema does not declare one', async () => { + const { emitter, publishedChannels } = emitterWith(SCHEMA) + + await emitter.emit([{ ...record('a'), channel: 'r/elsewhere' }]) + + expect(publishedChannels()).toEqual(['octolens-global']) + }) + + it('uses the sync unit channel for a passthrough schema record without one (dummy-style)', async () => { + const { emitter, publishedChannels } = emitterWith(z.record(z.unknown())) + + await emitter.emit([record('a')]) + + expect(publishedChannels()).toEqual(['octolens-global']) + }) + + it("keeps a record's own channel when the schema declares one", async () => { + const { emitter, publishedChannels } = emitterWith(SCHEMA_WITH_CHANNEL) + + await emitter.emit([ + { ...record('a'), channel: 'https://github.com/owner/repo' }, + { ...record('b'), channel: 'kubernetes' }, + record('c'), + { ...record('d'), channel: '' }, + ]) + + expect(publishedChannels()).toEqual([ + 'https://github.com/owner/repo', + 'kubernetes', + 'octolens-global', + 'octolens-global', + ]) + }) + + it("keeps a record's own channel on shadow records too", async () => { + const { deps, emitter } = emitterWith(SCHEMA_WITH_CHANNEL, false) + + await emitter.emit([{ ...record('a'), channel: 'kubernetes' }]) + + expect(deps.recordShadow).toHaveBeenCalledWith([ + expect.objectContaining({ data: expect.objectContaining({ channel: 'kubernetes' }) }), + ]) + }) + }) }) diff --git a/services/libs/connectors/src/emit.ts b/services/libs/connectors/src/emit.ts index e9c3f51de1..8d5a79dd0c 100644 --- a/services/libs/connectors/src/emit.ts +++ b/services/libs/connectors/src/emit.ts @@ -73,7 +73,7 @@ export function createEmit(deps: EmitterDeps): Emitter { throw err } - const payload = { ...parsed, channel: deps.unit.channelName } + const payload = { ...parsed, channel: recordChannel(parsed) ?? deps.unit.channelName } if (!deps.unit.emitEnabled) { const { type, sourceId, timestamp } = payload as Record @@ -120,3 +120,8 @@ export function createEmit(deps: EmitterDeps): Emitter { return { emit, emittedCount: () => emitted } } + +function recordChannel(parsed: Record): string | undefined { + const { channel } = parsed + return typeof channel === 'string' && channel.length > 0 ? channel : undefined +} diff --git a/services/libs/integrations/src/integrations/activityTypes.ts b/services/libs/integrations/src/integrations/activityTypes.ts index 0c458d318c..b6e5587a70 100644 --- a/services/libs/integrations/src/integrations/activityTypes.ts +++ b/services/libs/integrations/src/integrations/activityTypes.ts @@ -128,6 +128,15 @@ const defaultDiscourseFormatter = (activity) => { return `#${activity.channel}` } +const mentionActivityType = (platformLabel: string) => ({ + display: { + default: `mentioned you on ${platformLabel}`, + short: 'mentioned you', + channel: '', + }, + calculateSentiment: true, +}) + export const DEFAULT_ACTIVITY_TYPE_SETTINGS: DefaultActivityTypes = { [PlatformType.GITHUB]: { [GithubActivityType.DISCUSSION_STARTED]: { @@ -615,6 +624,7 @@ export const DEFAULT_ACTIVITY_TYPE_SETTINGS: DefaultActivityTypes = { }, calculateSentiment: true, }, + [DevToActivityType.MENTION]: mentionActivityType('DEV'), }, [PlatformType.DISCORD]: { [DiscordActivityType.JOINED_GUILD]: { @@ -687,6 +697,7 @@ export const DEFAULT_ACTIVITY_TYPE_SETTINGS: DefaultActivityTypes = { }, calculateSentiment: true, }, + [HackerNewsActivityType.MENTION]: mentionActivityType('Hacker News'), }, [PlatformType.LINKEDIN]: { [LinkedinActivityType.COMMENT]: { @@ -735,6 +746,7 @@ export const DEFAULT_ACTIVITY_TYPE_SETTINGS: DefaultActivityTypes = { }, calculateSentiment: true, }, + [RedditActivityType.MENTION]: mentionActivityType('Reddit'), }, [PlatformType.SLACK]: { [SlackActivityType.JOINED_CHANNEL]: { @@ -819,6 +831,7 @@ export const DEFAULT_ACTIVITY_TYPE_SETTINGS: DefaultActivityTypes = { }, calculateSentiment: true, }, + [StackOverflowActivityType.MENTION]: mentionActivityType('Stack Overflow'), }, [PlatformType.DISCOURSE]: { [DiscourseActivityType.CREATE_TOPIC]: { @@ -1518,4 +1531,25 @@ export const DEFAULT_ACTIVITY_TYPE_SETTINGS: DefaultActivityTypes = { calculateSentiment: false, }, }, + [PlatformType.MEDIUM]: { + mention: mentionActivityType('Medium'), + }, + [PlatformType.PRODUCTHUNT]: { + mention: mentionActivityType('Product Hunt'), + }, + [PlatformType.FACEBOOK]: { + mention: mentionActivityType('Facebook'), + }, + [PlatformType.YOUTUBE]: { + mention: mentionActivityType('YouTube'), + }, + [PlatformType.BLUESKY]: { + mention: mentionActivityType('Bluesky'), + }, + [PlatformType.TIKTOK]: { + mention: mentionActivityType('TikTok'), + }, + [PlatformType.INSTAGRAM]: { + mention: mentionActivityType('Instagram'), + }, } diff --git a/services/libs/integrations/src/integrations/devto/grid.ts b/services/libs/integrations/src/integrations/devto/grid.ts index f437abe30e..008eeb89ed 100644 --- a/services/libs/integrations/src/integrations/devto/grid.ts +++ b/services/libs/integrations/src/integrations/devto/grid.ts @@ -6,4 +6,7 @@ export const DEVTO_GRID: Record = { [DevToActivityType.COMMENT]: { score: 6, }, + [DevToActivityType.MENTION]: { + score: 6, + }, } diff --git a/services/libs/integrations/src/integrations/devto/types.ts b/services/libs/integrations/src/integrations/devto/types.ts index 02fce0f395..f9d3a8ad0d 100644 --- a/services/libs/integrations/src/integrations/devto/types.ts +++ b/services/libs/integrations/src/integrations/devto/types.ts @@ -13,6 +13,7 @@ export enum DevToRootStream { export enum DevToActivityType { COMMENT = 'comment', + MENTION = 'mention', } export interface IDevToRootOrganizationStreamData { diff --git a/services/libs/integrations/src/integrations/hackernews/grid.ts b/services/libs/integrations/src/integrations/hackernews/grid.ts index fc91dccac2..fc73f87ddb 100644 --- a/services/libs/integrations/src/integrations/hackernews/grid.ts +++ b/services/libs/integrations/src/integrations/hackernews/grid.ts @@ -9,4 +9,7 @@ export const HACKERNEWS_GRID: Record = { [RedditActivityType.COMMENT]: { score: 6, }, + [RedditActivityType.MENTION]: { + score: 6, + }, } diff --git a/services/libs/integrations/src/integrations/reddit/types.ts b/services/libs/integrations/src/integrations/reddit/types.ts index db205cc722..e87f4fa143 100644 --- a/services/libs/integrations/src/integrations/reddit/types.ts +++ b/services/libs/integrations/src/integrations/reddit/types.ts @@ -3,6 +3,7 @@ export const REDDIT_MAX_RETROSPECT_IN_HOURS = 5 export enum RedditActivityType { POST = 'post', COMMENT = 'comment', + MENTION = 'mention', } export { IRedditIntegrationSettings } from '@crowd/types' diff --git a/services/libs/integrations/src/integrations/stackoverflow/grid.ts b/services/libs/integrations/src/integrations/stackoverflow/grid.ts index 9d1d2c90cd..3564c5c6a4 100644 --- a/services/libs/integrations/src/integrations/stackoverflow/grid.ts +++ b/services/libs/integrations/src/integrations/stackoverflow/grid.ts @@ -9,4 +9,7 @@ export const STACKOVERFLOW_GRID: Record