feat: cio optimization

.infra/application.properties

@@ -27,5 +27,5 @@ debezium.transforms.ReadOperationFilter.language=jsr223.groovy
 debezium.transforms.ReadOperationFilter.condition=!(valueSchema.field('op') && value.op == 'r')
 debezium.transforms.PostsFilter.type=io.debezium.transforms.Filter
 debezium.transforms.PostsFilter.language=jsr223.groovy
-debezium.transforms.PostsFilter.condition=!(valueSchema.field('op') && value.op == 'u' && value.source.table == 'post' && value.before.views != value.after.views)
+debezium.transforms.PostsFilter.condition=!(valueSchema.field('op') && value.op == 'u' && value.source.table == 'post' && value.before.views != value.after.views) && !(valueSchema.field('op') && value.op == 'u' && value.source.table == 'user' && value.before.cioRegistered != value.after.cioRegistered && !value.after.cioRegistered)
 debezium.sink.type=pubsub

.infra/crons.ts

@@ -59,6 +59,10 @@ export const crons: Cron[] = [
     name: 'generic-referral-reminder',
     schedule: '12 3 * * *',
   },
+  // {
+  //   name: 'validate-active-users',
+  //   schedule: '15 4 * * *',
+  // },
   {
     name: 'update-source-tag-view',
     schedule: '20 3 * * 0',
@@ -109,5 +113,5 @@ export const crons: Cron[] = [
   {
     name: 'calculate-top-readers',
     schedule: '0 2 1 * *',
-  }
+  },
 ];

__tests__/cron/validateActiveUsers.ts

@@ -0,0 +1,249 @@
+import { crons } from '../../src/cron/index';
+import cron from '../../src/cron/validateActiveUsers';
+import { expectSuccessfulCron, saveFixtures } from '../helpers';
+import * as gcp from '../../src/common/googleCloud';
+import * as cioModule from '../../src/cio';
+import { DataSource, In, JsonContains } from 'typeorm';
+import createOrGetConnection from '../../src/db';
+import {
+  User,
+  UserPersonalizedDigest,
+  UserPersonalizedDigestSendType,
+} from '../../src/entity';
+import { badUsersFixture, plusUsersFixture, usersFixture } from '../fixture';
+import { updateFlagsStatement } from '../../src/common';
+import { ioRedisPool } from '../../src/redis';
+
+let con: DataSource;
+
+beforeEach(async () => {
+  con = await createOrGetConnection();
+  await ioRedisPool.execute((client) => client.flushall());
+  jest.clearAllMocks();
+  jest.resetModules();
+  await saveFixtures(con, User, [
+    ...usersFixture,
+    ...plusUsersFixture,
+    ...badUsersFixture,
+  ]);
+});
+
+describe('validateActiveUsers', () => {
+  beforeEach(async () => {
+    await saveFixtures(con, User, usersFixture);
+  });
+
+  it('should NOT be registered yet', () => {
+    const registeredWorker = crons.find((item) => item.name === cron.name);
+
+    expect(registeredWorker).toBeDefined();
+  });
+});
+
+describe('users for downgrade', () => {
+  it('should not do anything if users do not have digest subscription', async () => {
+    jest.spyOn(gcp, 'getUsersActiveState').mockResolvedValue({
+      reactivateUsers: [],
+      inactiveUsers: [],
+      downgradeUsers: ['4', '1'],
+    });
+
+    await con.getRepository(UserPersonalizedDigest).delete({});
+    await expectSuccessfulCron(cron);
+
+    const digests = await con.getRepository(UserPersonalizedDigest).find();
+    expect(digests.length).toEqual(0);
+  });
+
+  it('should downgrade daily digest to weekly digest', async () => {
+    const downgradeUsers = ['4', '1'];
+
+    jest.spyOn(gcp, 'getUsersActiveState').mockResolvedValue({
+      reactivateUsers: [],
+      inactiveUsers: [],
+      downgradeUsers,
+    });
+
+    await con.getRepository(UserPersonalizedDigest).update(
+      {},
+      {
+        preferredDay: 1,
+        preferredHour: 4,
+        flags: updateFlagsStatement({
+          digestSendType: UserPersonalizedDigestSendType.workdays,
+        }),
+      },
+    );
+    await expectSuccessfulCron(cron);
+
+    const digests = await con.getRepository(UserPersonalizedDigest).find({
+      where: {
+        flags: JsonContains({
+          digestSendType: UserPersonalizedDigestSendType.weekly,
+        }),
+      },
+    });
+    const downgradedOnly = digests.every(
+      ({ userId, preferredDay, preferredHour }) =>
+        downgradeUsers.includes(userId) &&
+        preferredDay === 3 &&
+        preferredHour === 9,
+    );
+    expect(downgradedOnly).toEqual(true);
+  });
+});
+
+describe('users for removal', () => {
+  it('should not do anything if users are removed to CIO already', async () => {
+    jest.spyOn(gcp, 'getUsersActiveState').mockResolvedValue({
+      reactivateUsers: ['1', '2'],
+      inactiveUsers: ['3', '5'],
+      downgradeUsers: ['4'],
+    });
+
+    const postSpy = jest
+      .spyOn(cioModule.cioV2.request, 'post')
+      .mockResolvedValue({});
+
+    await con
+      .getRepository(User)
+      .update({ id: In(['3', '5']) }, { cioRegistered: false });
+
+    await expectSuccessfulCron(cron);
+
+    expect(postSpy).not.toHaveBeenCalled();
+  });
+
+  it('should send removal to cio', async () => {
+    jest.spyOn(gcp, 'getUsersActiveState').mockResolvedValue({
+      reactivateUsers: ['1', '2'],
+      inactiveUsers: ['3', '5', 'vordr'],
+      downgradeUsers: ['4'],
+    });
+
+    const postSpy = jest
+      .spyOn(cioModule.cioV2.request, 'post')
+      .mockResolvedValue({});
+
+    await con
+      .getRepository(User)
+      .update({ id: In(['vordr']) }, { cioRegistered: false });
+    const digests = await con.getRepository(UserPersonalizedDigest).count();
+    expect(digests).toBeGreaterThan(0);
+
+    await expectSuccessfulCron(cron);
+
+    const batch = [
+      {
+        action: 'destroy',
+        type: 'person',
+        identifiers: { id: '3' },
+      },
+      {
+        action: 'destroy',
+        type: 'person',
+        identifiers: { id: '5' },
+      },
+    ];
+
+    expect(postSpy).toHaveBeenCalledWith('/users', { batch });
+
+    const fromRemovalOnly = postSpy.mock.calls[0][1].batch.every(
+      ({ identifiers }) => ['3', '5'].includes(identifiers.id),
+    );
+    expect(fromRemovalOnly).toBeTruthy();
+
+    const unRegisteredOnly = postSpy.mock.calls[0][1].batch.every(
+      ({ identifiers }) => !['vordr'].includes(identifiers.id),
+    );
+    expect(unRegisteredOnly).toBeTruthy();
+
+    const unregistered = await con
+      .getRepository(User)
+      .findOne({ select: ['cioRegistered'], where: { id: '3' } });
+    expect(unregistered.cioRegistered).toEqual(false);
+
+    const removed = await con.getRepository(UserPersonalizedDigest).count();
+    expect(removed).toBeLessThan(digests);
+  });
+});
+
+describe('users for reactivation', () => {
+  it('should not do anything if users are registered to CIO already', async () => {
+    jest.spyOn(gcp, 'getUsersActiveState').mockResolvedValue({
+      reactivateUsers: ['1', '2'],
+      inactiveUsers: ['3'],
+      downgradeUsers: ['4'],
+    });
+    const postSpy = jest
+      .spyOn(cioModule.cioV2.request, 'post')
+      .mockResolvedValue({});
+
+    // to stop running removal of `inactiveUsers`
+    await con.getRepository(User).update({ id: '3' }, { cioRegistered: false });
+
+    // default value for `cioRegistered` is true
+    await expectSuccessfulCron(cron);
+
+    expect(postSpy).not.toHaveBeenCalled();
+  });
+
+  it('should send reactivation to cio', async () => {
+    jest.spyOn(gcp, 'getUsersActiveState').mockResolvedValue({
+      reactivateUsers: ['1', '2'],
+      inactiveUsers: ['3'],
+      downgradeUsers: ['4'],
+    });
+
+    const postSpy = jest
+      .spyOn(cioModule.cioV2.request, 'post')
+      .mockResolvedValue({});
+
+    // to stop running removal of `inactiveUsers`
+    await con
+      .getRepository(User)
+      .update({ id: In(['3', '1']) }, { cioRegistered: false });
+
+    // default value for `cioRegistered` is true
+    await expectSuccessfulCron(cron);
+
+    const batch = [
+      {
+        action: 'identify',
+        type: 'person',
+        identifiers: { id: '1' },
+        attributes: {
+          accepted_marketing: false,
+          'cio_subscription_preferences.topics.topic_4': false,
+          'cio_subscription_preferences.topics.topic_7': true,
+          'cio_subscription_preferences.topics.topic_8': true,
+          'cio_subscription_preferences.topics.topic_9': true,
+          created_at: 1656427727,
+          first_name: 'Ido',
+          name: 'Ido',
+          permalink: 'http://localhost:5002/idoshamun',
+          referral_link: 'http://localhost:5002/join?cid=generic&userid=1',
+          updated_at: undefined,
+          username: 'idoshamun',
+        },
+      },
+    ];
+
+    expect(postSpy).toHaveBeenCalledWith('/users', { batch });
+
+    const fromReactivateUserOnly = postSpy.mock.calls[0][1].batch.every(
+      ({ identifiers }) => ['1'].includes(identifiers.id),
+    );
+    expect(fromReactivateUserOnly).toBeTruthy();
+
+    const registeredOnly = postSpy.mock.calls[0][1].batch.every(
+      ({ identifiers }) => identifiers.id !== '3',
+    );
+    expect(registeredOnly).toBeTruthy();
+
+    const reactivated = await con
+      .getRepository(User)
+      .findOne({ select: ['cioRegistered'], where: { id: '1' } });
+    expect(reactivated.cioRegistered).toEqual(true);
+  });
+});

__tests__/workers/userCreatedCio.ts

@@ -12,8 +12,8 @@ import { cio } from '../../src/cio';
 import { typedWorkers } from '../../src/workers';
 import mocked = jest.mocked;
 
-jest.mock('../../src/common', () => ({
-  ...jest.requireActual('../../src/common'),
+jest.mock('../../src/common/links', () => ({
+  ...jest.requireActual('../../src/common/links'),
   getShortGenericInviteLink: jest.fn(),
 }));
 

__tests__/workers/userUpdatedCio.ts

@@ -21,6 +21,11 @@ import { usersFixture } from '../fixture/user';
 
 jest.mock('../../src/common', () => ({
   ...jest.requireActual('../../src/common'),
+  resubscribeUser: jest.fn(),
+}));
+
+jest.mock('../../src/common/links', () => ({
+  ...jest.requireActual('../../src/common/links'),
   getShortGenericInviteLink: jest.fn(),
   resubscribeUser: jest.fn(),
 }));

bin/cioSyncBasedOnActivity.ts

@@ -0,0 +1,65 @@
+import createOrGetConnection from '../src/db';
+import fs from 'fs';
+import { parse } from 'csv-parse';
+import { syncSubscriptionsWithActiveState } from '../src/common';
+import {
+  GetUsersActiveState,
+  UserActiveState,
+} from '../src/common/googleCloud';
+
+const func = async () => {
+  const csvFilePath = process.argv[2];
+
+  if (!csvFilePath) {
+    throw new Error('CSV file path is required');
+  }
+
+  const users: GetUsersActiveState = {
+    inactiveUsers: [],
+    downgradeUsers: [],
+    reactivateUsers: [],
+  };
+
+  const stream = fs
+    .createReadStream(csvFilePath)
+    .pipe(parse({ delimiter: ',', from_line: 2 }));
+
+  stream.on('error', (err) => {
+    console.error('failed to read file: ', err.message);
+  });
+
+  stream.on('data', function ([id, rawStatus]) {
+    if (!id || !rawStatus) {
+      return;
+    }
+
+    const status = rawStatus.toString() as UserActiveState;
+
+    if (status === UserActiveState.InactiveSince6wAgo) {
+      users.downgradeUsers.push(id);
+    }
+
+    if (
+      status === UserActiveState.InactiveSince12wAgo ||
+      status === UserActiveState.NeverActive
+    ) {
+      users.inactiveUsers.push(id);
+    }
+  });
+
+  await new Promise((resolve) => {
+    stream.on('end', resolve);
+  });
+
+  console.log('running cron sync function');
+
+  const con = await createOrGetConnection();
+
+  await syncSubscriptionsWithActiveState({ con, users });
+
+  console.log('finished sync');
+
+  process.exit(0);
+};
+
+func();