From 81e5b5aeac0cc6fc0923dfbd90a08607724de47b Mon Sep 17 00:00:00 2001 From: capJavert Date: Thu, 23 Jul 2026 16:53:33 +0200 Subject: [PATCH 1/2] feat(personal-context): generate & persist user personal context Write-only integration with the personal-context-engine (PCE): - CDC on user-updated (portfolio change) triggers the website source - better-auth account.create.after hook triggers the github source, resolving the login via the GitHub API from the linked ba_account - shared requestPersonalContext helper upserts a pending row and publishes api.v1.generate-personal-context in one transaction - result worker persists profileText + boost/mute tags, guarded by a per-(userId,source) correlationId stale check - gated behind the opt-in personalContextEnabled remote config flag Co-Authored-By: Claude Opus 4.8 (1M context) --- .infra/common.ts | 12 +++ __tests__/setup.ts | 1 + .../personalContext/githubAccountLinked.ts | 76 +++++++++++++ .../personalContextGenerated.ts | 101 ++++++++++++++++++ .../userUpdatedPersonalContext.ts | 94 ++++++++++++++++ src/betterAuth.ts | 11 ++ .../personalContext/requestPersonalContext.ts | 52 +++++++++ src/common/schema/personalContext.ts | 40 +++++++ src/common/typedPubsub.ts | 13 +++ src/entity/user/UserPersonalContext.ts | 75 +++++++++++++ src/integrations/github/clients.ts | 26 ++++- src/integrations/github/types.ts | 5 + .../1784811495083-UserPersonalContext.ts | 27 +++++ src/remoteConfig.ts | 1 + src/workers/index.ts | 6 ++ .../personalContext/githubAccountLinked.ts | 40 +++++++ .../personalContextGenerated.ts | 45 ++++++++ .../userUpdatedPersonalContext.ts | 30 ++++++ 18 files changed, 654 insertions(+), 1 deletion(-) create mode 100644 __tests__/workers/personalContext/githubAccountLinked.ts create mode 100644 __tests__/workers/personalContext/personalContextGenerated.ts create mode 100644 __tests__/workers/personalContext/userUpdatedPersonalContext.ts create mode 100644 src/common/personalContext/requestPersonalContext.ts create mode 100644 src/common/schema/personalContext.ts create mode 100644 src/entity/user/UserPersonalContext.ts create mode 100644 src/migration/1784811495083-UserPersonalContext.ts create mode 100644 src/workers/personalContext/githubAccountLinked.ts create mode 100644 src/workers/personalContext/personalContextGenerated.ts create mode 100644 src/workers/personalContext/userUpdatedPersonalContext.ts diff --git a/.infra/common.ts b/.infra/common.ts index 0cbc030b13..b4a4f30d87 100644 --- a/.infra/common.ts +++ b/.infra/common.ts @@ -37,6 +37,14 @@ export const workers: Worker[] = [ topic: 'user-updated', subscription: 'api.user-updated-plus-subscribed-squad', }, + { + topic: 'user-updated', + subscription: 'api.user-updated-personal-context', + }, + { + topic: 'api.v1.github-account-linked', + subscription: 'api.github-account-linked-personal-context', + }, { topic: 'user-updated', subscription: 'api.user-updated-plus-subscribed-custom-feed', @@ -427,6 +435,10 @@ export const workers: Worker[] = [ topic: 'post-upvoted', subscription: 'api.post-upvoted-interest-signal', }, + { + topic: 'api.v1.personal-context-generated', + subscription: 'api.personal-context-generated', + }, { topic: 'api.v1.interest-content-available', subscription: 'api.interest-content-available-notification', diff --git a/__tests__/setup.ts b/__tests__/setup.ts index d2ab361f7a..205c51e871 100644 --- a/__tests__/setup.ts +++ b/__tests__/setup.ts @@ -30,6 +30,7 @@ jest.mock('../src/remoteConfig', () => ({ ignoredWorkEmailDomains: ['igored.com', 'ignored.org'], rateLimitReputationThreshold: 1, headlineChannelMinPosts: 3, + personalContextEnabled: true, pricingIds: { pricingGift: 'yearly' }, fees: { transfer: 5, diff --git a/__tests__/workers/personalContext/githubAccountLinked.ts b/__tests__/workers/personalContext/githubAccountLinked.ts new file mode 100644 index 0000000000..ad6a34bd1c --- /dev/null +++ b/__tests__/workers/personalContext/githubAccountLinked.ts @@ -0,0 +1,76 @@ +import nock from 'nock'; +import { DataSource } from 'typeorm'; +import { githubAccountLinkedWorker } from '../../../src/workers/personalContext/githubAccountLinked'; +import { triggerTypedEvent } from '../../../src/common/typedPubsub'; +import { expectSuccessfulTypedBackground, saveFixtures } from '../../helpers'; +import { User } from '../../../src/entity'; +import { + PersonalContextSource, + PersonalContextStatus, + UserPersonalContext, +} from '../../../src/entity/user/UserPersonalContext'; +import { usersFixture } from '../../fixture/user'; +import createOrGetConnection from '../../../src/db'; + +jest.mock('../../../src/common/typedPubsub', () => ({ + ...jest.requireActual('../../../src/common/typedPubsub'), + triggerTypedEvent: jest.fn(), +})); + +let con: DataSource; + +beforeAll(async () => { + con = await createOrGetConnection(); +}); + +beforeEach(async () => { + jest.clearAllMocks(); + nock.cleanAll(); + await saveFixtures(con, User, usersFixture); + await con.query(`DELETE FROM ba_account`); +}); + +const linkGithubAccount = () => + con.query( + `INSERT INTO ba_account (id, "accountId", "providerId", "userId", "accessToken") VALUES ('acc-1', '123', 'github', '1', 'gh-token')`, + ); + +describe('githubAccountLinked', () => { + it('resolves the github login and requests context for a linked account', async () => { + await linkGithubAccount(); + nock('https://api.github.com') + .get('/user') + .reply(200, { login: 'octocat' }); + + await expectSuccessfulTypedBackground(githubAccountLinkedWorker, { + userId: '1', + }); + + const row = await con + .getRepository(UserPersonalContext) + .findOneBy({ userId: '1', source: PersonalContextSource.Github }); + expect(row).toMatchObject({ + sourceValue: 'octocat', + verified: true, + status: PersonalContextStatus.Pending, + }); + expect(triggerTypedEvent).toHaveBeenCalledWith( + expect.anything(), + 'api.v1.generate-personal-context', + expect.objectContaining({ + userId: '1', + sources: [{ kind: PersonalContextSource.Github, value: 'octocat' }], + }), + ); + }); + + it('does nothing when the user has no linked github account', async () => { + await expectSuccessfulTypedBackground(githubAccountLinkedWorker, { + userId: '1', + }); + + const count = await con.getRepository(UserPersonalContext).count(); + expect(count).toBe(0); + expect(triggerTypedEvent).not.toHaveBeenCalled(); + }); +}); diff --git a/__tests__/workers/personalContext/personalContextGenerated.ts b/__tests__/workers/personalContext/personalContextGenerated.ts new file mode 100644 index 0000000000..d6e83e3f1d --- /dev/null +++ b/__tests__/workers/personalContext/personalContextGenerated.ts @@ -0,0 +1,101 @@ +import { DataSource } from 'typeorm'; +import { personalContextGeneratedWorker } from '../../../src/workers/personalContext/personalContextGenerated'; +import { expectSuccessfulTypedBackground, saveFixtures } from '../../helpers'; +import { User } from '../../../src/entity'; +import { + PersonalContextSource, + PersonalContextStatus, + UserPersonalContext, +} from '../../../src/entity/user/UserPersonalContext'; +import { usersFixture } from '../../fixture/user'; +import createOrGetConnection from '../../../src/db'; + +let con: DataSource; + +beforeAll(async () => { + con = await createOrGetConnection(); +}); + +beforeEach(async () => { + await saveFixtures(con, User, usersFixture); + await con.getRepository(UserPersonalContext).save({ + userId: '1', + source: PersonalContextSource.Website, + sourceValue: 'https://new.dev', + verified: true, + status: PersonalContextStatus.Pending, + correlationId: 'corr-1', + requestedAt: new Date(), + }); +}); + +const getRow = () => + con + .getRepository(UserPersonalContext) + .findOneBy({ userId: '1', source: PersonalContextSource.Website }); + +describe('personalContextGenerated', () => { + it('stores the synthesized context on a matching correlationId', async () => { + await expectSuccessfulTypedBackground(personalContextGeneratedWorker, { + userId: '1', + correlationId: 'corr-1', + status: 'ok', + profileText: 'A backend engineer who loves Go.', + context: { + profile_text: 'A backend engineer who loves Go.', + ranking_signals: { boost_tags: ['go', 'rust'], mute_tags: ['php'] }, + }, + }); + + const row = await getRow(); + expect(row).toMatchObject({ + status: PersonalContextStatus.Ok, + profileText: 'A backend engineer who loves Go.', + boostTags: ['go', 'rust'], + muteTags: ['php'], + }); + expect(row?.generatedAt).toBeTruthy(); + expect(row?.context).toMatchObject({ + ranking_signals: { boost_tags: ['go', 'rust'], mute_tags: ['php'] }, + }); + }); + + it('ignores a stale response whose correlationId no longer matches', async () => { + await expectSuccessfulTypedBackground(personalContextGeneratedWorker, { + userId: '1', + correlationId: 'stale', + status: 'ok', + profileText: 'should not be stored', + context: { ranking_signals: { boost_tags: ['x'], mute_tags: [] } }, + }); + + const row = await getRow(); + expect(row).toMatchObject({ + status: PersonalContextStatus.Pending, + profileText: null, + }); + }); + + it('records an error status and preserves prior profile text', async () => { + await con + .getRepository(UserPersonalContext) + .update( + { userId: '1', source: PersonalContextSource.Website }, + { status: PersonalContextStatus.Ok, profileText: 'previous good text' }, + ); + + await expectSuccessfulTypedBackground(personalContextGeneratedWorker, { + userId: '1', + correlationId: 'corr-1', + status: 'error', + error: 'could not read source', + }); + + const row = await getRow(); + expect(row).toMatchObject({ + status: PersonalContextStatus.Error, + error: 'could not read source', + profileText: 'previous good text', + }); + }); +}); diff --git a/__tests__/workers/personalContext/userUpdatedPersonalContext.ts b/__tests__/workers/personalContext/userUpdatedPersonalContext.ts new file mode 100644 index 0000000000..1059e4fe16 --- /dev/null +++ b/__tests__/workers/personalContext/userUpdatedPersonalContext.ts @@ -0,0 +1,94 @@ +import { DataSource } from 'typeorm'; +import { userUpdatedPersonalContextWorker } from '../../../src/workers/personalContext/userUpdatedPersonalContext'; +import { triggerTypedEvent } from '../../../src/common/typedPubsub'; +import { ChangeObject } from '../../../src/types'; +import { expectSuccessfulTypedBackground, saveFixtures } from '../../helpers'; +import { User } from '../../../src/entity'; +import { + PersonalContextSource, + PersonalContextStatus, + UserPersonalContext, +} from '../../../src/entity/user/UserPersonalContext'; +import { usersFixture } from '../../fixture/user'; +import createOrGetConnection from '../../../src/db'; + +jest.mock('../../../src/common/typedPubsub', () => ({ + ...jest.requireActual('../../../src/common/typedPubsub'), + triggerTypedEvent: jest.fn(), +})); + +let con: DataSource; + +beforeAll(async () => { + con = await createOrGetConnection(); +}); + +beforeEach(async () => { + jest.clearAllMocks(); + await saveFixtures(con, User, usersFixture); +}); + +const base = { + id: '1', + portfolio: 'https://old.dev', +} as ChangeObject; + +const invoke = ( + oldProfile: Partial>, + newProfile: Partial>, +) => + expectSuccessfulTypedBackground(userUpdatedPersonalContextWorker, { + user: { ...base, ...oldProfile } as ChangeObject, + newProfile: { ...base, ...newProfile } as ChangeObject, + }); + +describe('userUpdatedPersonalContext', () => { + it('requests context and stores a pending website row when portfolio changes', async () => { + await invoke( + { portfolio: 'https://old.dev' }, + { portfolio: 'https://new.dev' }, + ); + + const row = await con + .getRepository(UserPersonalContext) + .findOneBy({ userId: '1', source: PersonalContextSource.Website }); + + expect(row).toMatchObject({ + userId: '1', + source: PersonalContextSource.Website, + sourceValue: 'https://new.dev', + verified: true, + status: PersonalContextStatus.Pending, + }); + expect(row?.correlationId).toBeTruthy(); + expect(triggerTypedEvent).toHaveBeenCalledWith( + expect.anything(), + 'api.v1.generate-personal-context', + expect.objectContaining({ + userId: '1', + sources: [ + { kind: PersonalContextSource.Website, value: 'https://new.dev' }, + ], + }), + ); + }); + + it('does nothing when portfolio is unchanged', async () => { + await invoke( + { portfolio: 'https://same.dev', name: 'old' }, + { portfolio: 'https://same.dev', name: 'new' }, + ); + + const count = await con.getRepository(UserPersonalContext).count(); + expect(count).toBe(0); + expect(triggerTypedEvent).not.toHaveBeenCalled(); + }); + + it('does nothing when the new portfolio is empty', async () => { + await invoke({ portfolio: 'https://old.dev' }, { portfolio: '' }); + + const count = await con.getRepository(UserPersonalContext).count(); + expect(count).toBe(0); + expect(triggerTypedEvent).not.toHaveBeenCalled(); + }); +}); diff --git a/src/betterAuth.ts b/src/betterAuth.ts index 8383364907..5c197385f4 100644 --- a/src/betterAuth.ts +++ b/src/betterAuth.ts @@ -777,6 +777,17 @@ export const getBetterAuthOptions = (pool: Pool): BetterAuthOptions => { }, }, }, + account: { + create: { + after: async (account) => { + if (account.providerId === 'github') { + await triggerTypedEvent(logger, 'api.v1.github-account-linked', { + userId: account.userId, + }); + } + }, + }, + }, }, emailAndPassword: { enabled: true, diff --git a/src/common/personalContext/requestPersonalContext.ts b/src/common/personalContext/requestPersonalContext.ts new file mode 100644 index 0000000000..29eb4933b3 --- /dev/null +++ b/src/common/personalContext/requestPersonalContext.ts @@ -0,0 +1,52 @@ +import type { DataSource, EntityManager } from 'typeorm'; +import { + PersonalContextSource, + PersonalContextStatus, + UserPersonalContext, +} from '../../entity/user/UserPersonalContext'; +import { generateShortId } from '../../ids'; +import { triggerTypedEvent } from '../typedPubsub'; +import { logger } from '../../logger'; + +export const requestPersonalContext = async ({ + con, + userId, + source, + value, + verified, +}: { + con: DataSource | EntityManager; + userId: string; + source: PersonalContextSource; + value?: string | null; + verified: boolean; +}): Promise => { + if (!value) { + return false; + } + + const correlationId = await generateShortId(); + + await con.transaction(async (manager) => { + await manager.getRepository(UserPersonalContext).upsert( + { + userId, + source, + sourceValue: value, + verified, + status: PersonalContextStatus.Pending, + correlationId, + requestedAt: new Date(), + }, + ['userId', 'source'], + ); + + await triggerTypedEvent(logger, 'api.v1.generate-personal-context', { + userId, + correlationId, + sources: [{ kind: source, value }], + }); + }); + + return true; +}; diff --git a/src/common/schema/personalContext.ts b/src/common/schema/personalContext.ts new file mode 100644 index 0000000000..23b627c0d7 --- /dev/null +++ b/src/common/schema/personalContext.ts @@ -0,0 +1,40 @@ +import { z } from 'zod'; + +const personalContextSourceRefSchema = z.object({ + kind: z.string(), + value: z.string(), +}); + +export const generatePersonalContextSchema = z.object({ + userId: z.string(), + correlationId: z.string().optional(), + sources: z.array(personalContextSourceRefSchema), +}); + +const personalContextRankingSignalsSchema = z.object({ + boost_tags: z.array(z.string()).default([]), + mute_tags: z.array(z.string()).default([]), +}); + +const personalContextContextSchema = z + .object({ + headline: z.string().nullish(), + summary: z.string().nullish(), + seniority: z.string().nullish(), + current_focus: z.string().nullish(), + learning_goals: z.array(z.string()).nullish(), + ranking_signals: personalContextRankingSignalsSchema.nullish(), + profile_text: z.string().nullish(), + }) + .catchall(z.unknown()); + +export const personalContextGeneratedSchema = z.object({ + userId: z.string(), + correlationId: z.string().optional(), + status: z.enum(['ok', 'error']), + context: personalContextContextSchema.optional(), + profileText: z.string().optional(), + evidence: z.array(z.unknown()).optional(), + errors: z.array(z.unknown()).optional(), + error: z.string().optional(), +}); diff --git a/src/common/typedPubsub.ts b/src/common/typedPubsub.ts index 2663074fbe..ff1f3e0fc4 100644 --- a/src/common/typedPubsub.ts +++ b/src/common/typedPubsub.ts @@ -57,6 +57,10 @@ import type { opportunityFeedbackSchema, rejectionFeedbackClassificationSchema, } from './schema/opportunityMatch'; +import type { + generatePersonalContextSchema, + personalContextGeneratedSchema, +} from './schema/personalContext'; export type PubSubSchema = { 'pub-request': { @@ -343,6 +347,15 @@ export type PubSubSchema = { digestKey: string; scheduledAt: string; }; + 'api.v1.generate-personal-context': z.infer< + typeof generatePersonalContextSchema + >; + 'api.v1.personal-context-generated': z.infer< + typeof personalContextGeneratedSchema + >; + 'api.v1.github-account-linked': { + userId: string; + }; }; export async function triggerTypedEvent( diff --git a/src/entity/user/UserPersonalContext.ts b/src/entity/user/UserPersonalContext.ts new file mode 100644 index 0000000000..e648562205 --- /dev/null +++ b/src/entity/user/UserPersonalContext.ts @@ -0,0 +1,75 @@ +import { + Column, + CreateDateColumn, + Entity, + Index, + JoinColumn, + ManyToOne, + PrimaryColumn, + UpdateDateColumn, +} from 'typeorm'; +import type { User } from './User'; + +export enum PersonalContextSource { + Github = 'github', + Website = 'website', +} + +export enum PersonalContextStatus { + Pending = 'pending', + Ok = 'ok', + Error = 'error', +} + +@Entity() +@Index('IDX_user_personal_context_user_id', ['userId']) +export class UserPersonalContext { + @PrimaryColumn({ type: 'text' }) + userId: string; + + @PrimaryColumn({ type: 'text' }) + source: PersonalContextSource; + + @Column({ type: 'text' }) + sourceValue: string; + + @Column({ type: 'boolean', default: false }) + verified: boolean; + + @Column({ type: 'text', default: PersonalContextStatus.Pending }) + status: PersonalContextStatus; + + @Column({ type: 'text', nullable: true }) + profileText: string | null; + + @Column({ type: 'text', array: true, default: () => `'{}'` }) + boostTags: string[]; + + @Column({ type: 'text', array: true, default: () => `'{}'` }) + muteTags: string[]; + + @Column({ type: 'jsonb', nullable: true }) + context: Record | null; + + @Column({ type: 'text', nullable: true }) + error: string | null; + + @Column({ type: 'text', nullable: true }) + correlationId: string | null; + + @Column({ type: 'timestamptz', nullable: true }) + requestedAt: Date | null; + + @Column({ type: 'timestamptz', nullable: true }) + generatedAt: Date | null; + + @CreateDateColumn() + createdAt: Date; + + @UpdateDateColumn() + updatedAt: Date; + + @ManyToOne('User', { lazy: true, onDelete: 'CASCADE' }) + @JoinColumn({ name: 'userId' }) + user: Promise; +} diff --git a/src/integrations/github/clients.ts b/src/integrations/github/clients.ts index 37e45f815f..77fb3bca77 100644 --- a/src/integrations/github/clients.ts +++ b/src/integrations/github/clients.ts @@ -1,6 +1,10 @@ import fetch from 'node-fetch'; import { GarmrService, IGarmrService, GarmrNoopService } from '../garmr'; -import { IGitHubClient, GitHubSearchResponse } from './types'; +import { + IGitHubClient, + GitHubSearchResponse, + GitHubAuthenticatedUser, +} from './types'; export class GitHubClient implements IGitHubClient { private readonly baseUrl: string; @@ -46,6 +50,26 @@ export class GitHubClient implements IGitHubClient { return response.json() as Promise; }); } + + async getAuthenticatedUser(token: string): Promise { + return this.garmr.execute(async () => { + const response = await fetch(`${this.baseUrl}/user`, { + headers: { + Accept: 'application/vnd.github.v3+json', + 'User-Agent': 'daily.dev', + Authorization: `Bearer ${token}`, + }, + }); + + if (!response.ok) { + throw new Error( + `GitHub API error: ${response.status} ${response.statusText}`, + ); + } + + return response.json() as Promise; + }); + } } // Configure Garmr service for GitHub diff --git a/src/integrations/github/types.ts b/src/integrations/github/types.ts index acfb57b2c8..6a5c8d18ff 100644 --- a/src/integrations/github/types.ts +++ b/src/integrations/github/types.ts @@ -29,9 +29,14 @@ export interface GQLGitHubRepository { description: string | null; } +export interface GitHubAuthenticatedUser { + login: string; +} + export interface IGitHubClient extends IGarmrClient { searchRepositories( query: string, limit?: number, ): Promise; + getAuthenticatedUser(token: string): Promise; } diff --git a/src/migration/1784811495083-UserPersonalContext.ts b/src/migration/1784811495083-UserPersonalContext.ts new file mode 100644 index 0000000000..9fc856b7de --- /dev/null +++ b/src/migration/1784811495083-UserPersonalContext.ts @@ -0,0 +1,27 @@ +import { MigrationInterface, QueryRunner } from 'typeorm'; + +export class UserPersonalContext1784811495083 implements MigrationInterface { + name = 'UserPersonalContext1784811495083'; + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query( + `CREATE TABLE IF NOT EXISTS "user_personal_context" ("userId" character varying NOT NULL, "source" text NOT NULL, "sourceValue" text NOT NULL, "verified" boolean NOT NULL DEFAULT false, "status" text NOT NULL DEFAULT 'pending', "profileText" text, "boostTags" text array NOT NULL DEFAULT '{}', "muteTags" text array NOT NULL DEFAULT '{}', "context" jsonb, "error" text, "correlationId" text, "requestedAt" TIMESTAMP WITH TIME ZONE, "generatedAt" TIMESTAMP WITH TIME ZONE, "createdAt" TIMESTAMP NOT NULL DEFAULT now(), "updatedAt" TIMESTAMP NOT NULL DEFAULT now(), CONSTRAINT "PK_359a238fc279d413ede7eb473ee" PRIMARY KEY ("userId", "source"))`, + ); + await queryRunner.query( + `CREATE INDEX IF NOT EXISTS "IDX_user_personal_context_user_id" ON "user_personal_context" ("userId")`, + ); + await queryRunner.query( + `ALTER TABLE "user_personal_context" ADD CONSTRAINT "FK_8d89fb7a971d70503156013ba08" FOREIGN KEY ("userId") REFERENCES "user"("id") ON DELETE CASCADE ON UPDATE NO ACTION`, + ); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query( + `ALTER TABLE IF EXISTS "user_personal_context" DROP CONSTRAINT IF EXISTS "FK_8d89fb7a971d70503156013ba08"`, + ); + await queryRunner.query( + `DROP INDEX IF EXISTS "public"."IDX_user_personal_context_user_id"`, + ); + await queryRunner.query(`DROP TABLE IF EXISTS "user_personal_context"`); + } +} diff --git a/src/remoteConfig.ts b/src/remoteConfig.ts index 39fb3f3049..93fa6e8ae1 100644 --- a/src/remoteConfig.ts +++ b/src/remoteConfig.ts @@ -68,6 +68,7 @@ export type RemoteConfigValue = { contributionProgram: ContributionProgramConfig; headlineChannelMinPosts: number; excludedMarketingCta: string[]; + personalContextEnabled: boolean; }; class RemoteConfig { diff --git a/src/workers/index.ts b/src/workers/index.ts index 05aa14a624..d312d6e9b4 100644 --- a/src/workers/index.ts +++ b/src/workers/index.ts @@ -70,6 +70,9 @@ import { userInterestRunWorker } from './interest/userInterestRun'; import { postVisibleInterestMatchWorker } from './interest/postVisibleInterestMatch'; import { postUpvotedInterestSignalWorker } from './interest/postUpvotedInterestSignal'; import { userUpdatedPlusSubscriptionBriefWorker } from './userUpdatedPlusSubscriptionBrief'; +import { userUpdatedPersonalContextWorker } from './personalContext/userUpdatedPersonalContext'; +import { githubAccountLinkedWorker } from './personalContext/githubAccountLinked'; +import { personalContextGeneratedWorker } from './personalContext/personalContextGenerated'; import { postAddedSlackChannelSendBriefWorker } from './postAddedSlackChannelSendBrief'; import campaignUpdatedAction from './campaignUpdatedAction'; import campaignUpdatedSlack from './campaignUpdatedSlack'; @@ -165,6 +168,9 @@ export const typedWorkers: BaseTypedWorker[] = [ postVisibleInterestMatchWorker, postUpvotedInterestSignalWorker, userUpdatedPlusSubscriptionBriefWorker, + userUpdatedPersonalContextWorker, + githubAccountLinkedWorker, + personalContextGeneratedWorker, postAddedSlackChannelSendBriefWorker, postAnalyticsUpdate, postAuthorReputationEvent, diff --git a/src/workers/personalContext/githubAccountLinked.ts b/src/workers/personalContext/githubAccountLinked.ts new file mode 100644 index 0000000000..653fa89667 --- /dev/null +++ b/src/workers/personalContext/githubAccountLinked.ts @@ -0,0 +1,40 @@ +import { TypedWorker } from '../worker'; +import { PersonalContextSource } from '../../entity/user/UserPersonalContext'; +import { requestPersonalContext } from '../../common/personalContext/requestPersonalContext'; +import { gitHubClient } from '../../integrations/github/clients'; +import { remoteConfig } from '../../remoteConfig'; + +export const githubAccountLinkedWorker: TypedWorker<'api.v1.github-account-linked'> = + { + subscription: 'api.github-account-linked-personal-context', + handler: async (message, con) => { + const { + data: { userId }, + } = message; + + if (!remoteConfig.vars.personalContextEnabled) { + return; + } + + const [account] = await con.query( + `SELECT "accessToken" FROM ba_account WHERE "userId" = $1 AND "providerId" = 'github' LIMIT 1`, + [userId], + ); + + if (!account?.accessToken) { + return; + } + + const { login } = await gitHubClient.getAuthenticatedUser( + account.accessToken, + ); + + await requestPersonalContext({ + con, + userId, + source: PersonalContextSource.Github, + value: login, + verified: true, + }); + }, + }; diff --git a/src/workers/personalContext/personalContextGenerated.ts b/src/workers/personalContext/personalContextGenerated.ts new file mode 100644 index 0000000000..b1d0ca11bc --- /dev/null +++ b/src/workers/personalContext/personalContextGenerated.ts @@ -0,0 +1,45 @@ +import { TypedWorker } from '../worker'; +import { + PersonalContextStatus, + UserPersonalContext, +} from '../../entity/user/UserPersonalContext'; + +export const personalContextGeneratedWorker: TypedWorker<'api.v1.personal-context-generated'> = + { + subscription: 'api.personal-context-generated', + handler: async (message, con) => { + const { data } = message; + const { userId, correlationId, status } = data; + + if (!correlationId) { + return; + } + + const repo = con.getRepository(UserPersonalContext); + const row = await repo.findOneBy({ userId, correlationId }); + + if (!row) { + return; + } + + row.generatedAt = new Date(); + + if (status === 'error') { + row.status = PersonalContextStatus.Error; + row.error = data.error ?? 'unknown error'; + await repo.save(row); + return; + } + + const rankingSignals = data.context?.ranking_signals; + + row.status = PersonalContextStatus.Ok; + row.profileText = data.profileText ?? null; + row.boostTags = rankingSignals?.boost_tags ?? []; + row.muteTags = rankingSignals?.mute_tags ?? []; + row.context = data.context ?? null; + row.error = null; + + await repo.save(row); + }, + }; diff --git a/src/workers/personalContext/userUpdatedPersonalContext.ts b/src/workers/personalContext/userUpdatedPersonalContext.ts new file mode 100644 index 0000000000..2dfcf1cba2 --- /dev/null +++ b/src/workers/personalContext/userUpdatedPersonalContext.ts @@ -0,0 +1,30 @@ +import { TypedWorker } from '../worker'; +import { isChanged } from '../cdc/common'; +import { PersonalContextSource } from '../../entity/user/UserPersonalContext'; +import { requestPersonalContext } from '../../common/personalContext/requestPersonalContext'; +import { remoteConfig } from '../../remoteConfig'; + +export const userUpdatedPersonalContextWorker: TypedWorker<'user-updated'> = { + subscription: 'api.user-updated-personal-context', + handler: async (message, con) => { + const { + data: { newProfile, user: oldUser }, + } = message; + + if (!remoteConfig.vars.personalContextEnabled) { + return; + } + + if (!isChanged(oldUser, newProfile, 'portfolio')) { + return; + } + + await requestPersonalContext({ + con, + userId: newProfile.id, + source: PersonalContextSource.Website, + value: newProfile.portfolio, + verified: true, + }); + }, +}; From 19e3ac739f853d90abc9f44a98bf59da996e7d26 Mon Sep 17 00:00:00 2001 From: capJavert Date: Fri, 24 Jul 2026 12:22:06 +0200 Subject: [PATCH 2/2] refactor(personal-context): address review feedback - align UserPersonalContext.userId with user.id (varchar(36)) to avoid entity/migration drift - commit the pending row before publishing generate-personal-context (was inside a transaction) so PCE can't receive a request before the row exists; publish failure self-heals via trigger-message retry - drop the redundant userId index (composite PK already covers it) - remove dedicated boostTags/muteTags columns; tags live only in context.ranking_signals to avoid duplication/drift - githubAccountLinked: mark the source failed on a GitHub API error instead of nacking and retrying forever (revoked/expired token) Co-Authored-By: Claude Opus 4.8 (1M context) --- .../personalContext/githubAccountLinked.ts | 19 ++++++++++ .../personalContextGenerated.ts | 2 - .../personalContext/requestPersonalContext.ts | 32 ++++++++-------- src/entity/user/UserPersonalContext.ts | 10 +---- .../1784811495083-UserPersonalContext.ts | 8 +--- .../personalContext/githubAccountLinked.ts | 37 ++++++++++++++++--- .../personalContextGenerated.ts | 4 -- 7 files changed, 67 insertions(+), 45 deletions(-) diff --git a/__tests__/workers/personalContext/githubAccountLinked.ts b/__tests__/workers/personalContext/githubAccountLinked.ts index ad6a34bd1c..de9db4ff4f 100644 --- a/__tests__/workers/personalContext/githubAccountLinked.ts +++ b/__tests__/workers/personalContext/githubAccountLinked.ts @@ -73,4 +73,23 @@ describe('githubAccountLinked', () => { expect(count).toBe(0); expect(triggerTypedEvent).not.toHaveBeenCalled(); }); + + it('marks the source failed and does not publish on a github api error', async () => { + await linkGithubAccount(); + nock('https://api.github.com').get('/user').times(2).reply(401); + + await expectSuccessfulTypedBackground(githubAccountLinkedWorker, { + userId: '1', + }); + + const row = await con + .getRepository(UserPersonalContext) + .findOneBy({ userId: '1', source: PersonalContextSource.Github }); + expect(row).toMatchObject({ + sourceValue: '123', + status: PersonalContextStatus.Error, + }); + expect(row?.error).toBeTruthy(); + expect(triggerTypedEvent).not.toHaveBeenCalled(); + }); }); diff --git a/__tests__/workers/personalContext/personalContextGenerated.ts b/__tests__/workers/personalContext/personalContextGenerated.ts index d6e83e3f1d..7838af0986 100644 --- a/__tests__/workers/personalContext/personalContextGenerated.ts +++ b/__tests__/workers/personalContext/personalContextGenerated.ts @@ -51,8 +51,6 @@ describe('personalContextGenerated', () => { expect(row).toMatchObject({ status: PersonalContextStatus.Ok, profileText: 'A backend engineer who loves Go.', - boostTags: ['go', 'rust'], - muteTags: ['php'], }); expect(row?.generatedAt).toBeTruthy(); expect(row?.context).toMatchObject({ diff --git a/src/common/personalContext/requestPersonalContext.ts b/src/common/personalContext/requestPersonalContext.ts index 29eb4933b3..2d02851ada 100644 --- a/src/common/personalContext/requestPersonalContext.ts +++ b/src/common/personalContext/requestPersonalContext.ts @@ -27,25 +27,23 @@ export const requestPersonalContext = async ({ const correlationId = await generateShortId(); - await con.transaction(async (manager) => { - await manager.getRepository(UserPersonalContext).upsert( - { - userId, - source, - sourceValue: value, - verified, - status: PersonalContextStatus.Pending, - correlationId, - requestedAt: new Date(), - }, - ['userId', 'source'], - ); - - await triggerTypedEvent(logger, 'api.v1.generate-personal-context', { + await con.getRepository(UserPersonalContext).upsert( + { userId, + source, + sourceValue: value, + verified, + status: PersonalContextStatus.Pending, correlationId, - sources: [{ kind: source, value }], - }); + requestedAt: new Date(), + }, + ['userId', 'source'], + ); + + await triggerTypedEvent(logger, 'api.v1.generate-personal-context', { + userId, + correlationId, + sources: [{ kind: source, value }], }); return true; diff --git a/src/entity/user/UserPersonalContext.ts b/src/entity/user/UserPersonalContext.ts index e648562205..5ab774cfbf 100644 --- a/src/entity/user/UserPersonalContext.ts +++ b/src/entity/user/UserPersonalContext.ts @@ -2,7 +2,6 @@ import { Column, CreateDateColumn, Entity, - Index, JoinColumn, ManyToOne, PrimaryColumn, @@ -22,9 +21,8 @@ export enum PersonalContextStatus { } @Entity() -@Index('IDX_user_personal_context_user_id', ['userId']) export class UserPersonalContext { - @PrimaryColumn({ type: 'text' }) + @PrimaryColumn({ type: 'varchar', length: 36 }) userId: string; @PrimaryColumn({ type: 'text' }) @@ -42,12 +40,6 @@ export class UserPersonalContext { @Column({ type: 'text', nullable: true }) profileText: string | null; - @Column({ type: 'text', array: true, default: () => `'{}'` }) - boostTags: string[]; - - @Column({ type: 'text', array: true, default: () => `'{}'` }) - muteTags: string[]; - @Column({ type: 'jsonb', nullable: true }) context: Record | null; diff --git a/src/migration/1784811495083-UserPersonalContext.ts b/src/migration/1784811495083-UserPersonalContext.ts index 9fc856b7de..9d636a2d6b 100644 --- a/src/migration/1784811495083-UserPersonalContext.ts +++ b/src/migration/1784811495083-UserPersonalContext.ts @@ -5,10 +5,7 @@ export class UserPersonalContext1784811495083 implements MigrationInterface { public async up(queryRunner: QueryRunner): Promise { await queryRunner.query( - `CREATE TABLE IF NOT EXISTS "user_personal_context" ("userId" character varying NOT NULL, "source" text NOT NULL, "sourceValue" text NOT NULL, "verified" boolean NOT NULL DEFAULT false, "status" text NOT NULL DEFAULT 'pending', "profileText" text, "boostTags" text array NOT NULL DEFAULT '{}', "muteTags" text array NOT NULL DEFAULT '{}', "context" jsonb, "error" text, "correlationId" text, "requestedAt" TIMESTAMP WITH TIME ZONE, "generatedAt" TIMESTAMP WITH TIME ZONE, "createdAt" TIMESTAMP NOT NULL DEFAULT now(), "updatedAt" TIMESTAMP NOT NULL DEFAULT now(), CONSTRAINT "PK_359a238fc279d413ede7eb473ee" PRIMARY KEY ("userId", "source"))`, - ); - await queryRunner.query( - `CREATE INDEX IF NOT EXISTS "IDX_user_personal_context_user_id" ON "user_personal_context" ("userId")`, + `CREATE TABLE IF NOT EXISTS "user_personal_context" ("userId" character varying(36) NOT NULL, "source" text NOT NULL, "sourceValue" text NOT NULL, "verified" boolean NOT NULL DEFAULT false, "status" text NOT NULL DEFAULT 'pending', "profileText" text, "context" jsonb, "error" text, "correlationId" text, "requestedAt" TIMESTAMP WITH TIME ZONE, "generatedAt" TIMESTAMP WITH TIME ZONE, "createdAt" TIMESTAMP NOT NULL DEFAULT now(), "updatedAt" TIMESTAMP NOT NULL DEFAULT now(), CONSTRAINT "PK_359a238fc279d413ede7eb473ee" PRIMARY KEY ("userId", "source"))`, ); await queryRunner.query( `ALTER TABLE "user_personal_context" ADD CONSTRAINT "FK_8d89fb7a971d70503156013ba08" FOREIGN KEY ("userId") REFERENCES "user"("id") ON DELETE CASCADE ON UPDATE NO ACTION`, @@ -19,9 +16,6 @@ export class UserPersonalContext1784811495083 implements MigrationInterface { await queryRunner.query( `ALTER TABLE IF EXISTS "user_personal_context" DROP CONSTRAINT IF EXISTS "FK_8d89fb7a971d70503156013ba08"`, ); - await queryRunner.query( - `DROP INDEX IF EXISTS "public"."IDX_user_personal_context_user_id"`, - ); await queryRunner.query(`DROP TABLE IF EXISTS "user_personal_context"`); } } diff --git a/src/workers/personalContext/githubAccountLinked.ts b/src/workers/personalContext/githubAccountLinked.ts index 653fa89667..f71bc91229 100644 --- a/src/workers/personalContext/githubAccountLinked.ts +++ b/src/workers/personalContext/githubAccountLinked.ts @@ -1,5 +1,9 @@ import { TypedWorker } from '../worker'; -import { PersonalContextSource } from '../../entity/user/UserPersonalContext'; +import { + PersonalContextSource, + PersonalContextStatus, + UserPersonalContext, +} from '../../entity/user/UserPersonalContext'; import { requestPersonalContext } from '../../common/personalContext/requestPersonalContext'; import { gitHubClient } from '../../integrations/github/clients'; import { remoteConfig } from '../../remoteConfig'; @@ -7,7 +11,7 @@ import { remoteConfig } from '../../remoteConfig'; export const githubAccountLinkedWorker: TypedWorker<'api.v1.github-account-linked'> = { subscription: 'api.github-account-linked-personal-context', - handler: async (message, con) => { + handler: async (message, con, logger) => { const { data: { userId }, } = message; @@ -17,7 +21,7 @@ export const githubAccountLinkedWorker: TypedWorker<'api.v1.github-account-linke } const [account] = await con.query( - `SELECT "accessToken" FROM ba_account WHERE "userId" = $1 AND "providerId" = 'github' LIMIT 1`, + `SELECT "accountId", "accessToken" FROM ba_account WHERE "userId" = $1 AND "providerId" = 'github' LIMIT 1`, [userId], ); @@ -25,9 +29,30 @@ export const githubAccountLinkedWorker: TypedWorker<'api.v1.github-account-linke return; } - const { login } = await gitHubClient.getAuthenticatedUser( - account.accessToken, - ); + let login: string; + try { + ({ login } = await gitHubClient.getAuthenticatedUser( + account.accessToken, + )); + } catch (err) { + logger.error( + { err, userId, provider: 'personal context' }, + 'failed to resolve github login for personal context', + ); + await con.getRepository(UserPersonalContext).upsert( + { + userId, + source: PersonalContextSource.Github, + sourceValue: account.accountId, + verified: true, + status: PersonalContextStatus.Error, + error: err instanceof Error ? err.message : String(err), + generatedAt: new Date(), + }, + ['userId', 'source'], + ); + return; + } await requestPersonalContext({ con, diff --git a/src/workers/personalContext/personalContextGenerated.ts b/src/workers/personalContext/personalContextGenerated.ts index b1d0ca11bc..da0cf7271e 100644 --- a/src/workers/personalContext/personalContextGenerated.ts +++ b/src/workers/personalContext/personalContextGenerated.ts @@ -31,12 +31,8 @@ export const personalContextGeneratedWorker: TypedWorker<'api.v1.personal-contex return; } - const rankingSignals = data.context?.ranking_signals; - row.status = PersonalContextStatus.Ok; row.profileText = data.profileText ?? null; - row.boostTags = rankingSignals?.boost_tags ?? []; - row.muteTags = rankingSignals?.mute_tags ?? []; row.context = data.context ?? null; row.error = null;