feat(personal-context): generate & persist user personal context

.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',

__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,

__tests__/workers/personalContext/githubAccountLinked.ts

@@ -0,0 +1,95 @@
+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();
+  });
+
+  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();
+  });
+});

__tests__/workers/personalContext/personalContextGenerated.ts

@@ -0,0 +1,99 @@
+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.',
+    });
+    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',
+    });
+  });
+});

__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<User>;
+
+const invoke = (
+  oldProfile: Partial<ChangeObject<User>>,
+  newProfile: Partial<ChangeObject<User>>,
+) =>
+  expectSuccessfulTypedBackground(userUpdatedPersonalContextWorker, {
+    user: { ...base, ...oldProfile } as ChangeObject<User>,
+    newProfile: { ...base, ...newProfile } as ChangeObject<User>,
+  });
+
+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();
+  });
+});

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,

src/common/personalContext/requestPersonalContext.ts

@@ -0,0 +1,50 @@
+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<boolean> => {
+  if (!value) {
+    return false;
+  }
+
+  const correlationId = await generateShortId();
+
+  await con.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;
+};