10411869195

Committed 15 Aug 2024 11:14PM UTC coverage: 32.239% (+0.2%) from 32.02%

Build # 10411869195

Build Type

Pull #9584

github

Committed by

web-flow

Commit Message

Merge a756cd301 into f6b991e92

Pull Request Pull Request #9584: Merge master into LW Deploy

Run Details

5015 of 40241 branches covered (12.46%)

Branch coverage included in aggregate %.

211 of 395 new or added lines in 72 files covered. (53.42%)

17 existing lines in 15 files now uncovered.

28958 of 65136 relevant lines covered (44.46%)

190.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

63.28

/packages/lesswrong/server/useractivities/cron.ts

/* eslint-disable no-console */
/* See lib/collections/useractivities/collection.ts for a high-level overview */
import chunk from 'lodash/fp/chunk';
import max from 'lodash/fp/max';
import { isEAForum, isLW } from '../../lib/instanceSettings';
import { randomId } from '../../lib/random';
import { getSqlClientOrThrow } from '@/server/sql/sqlClient';
import { addCronJob } from '../cronUtil';
import { Vulcan } from '../vulcan-lib';
import { ActivityWindowData, getUserActivityData } from './getUserActivityData';
import { isAnyTest } from '../../lib/executionEnvironment';

const ACTIVITY_WINDOW_HOURS = 21 * 24; // 3 weeks

/**
 * Assert that all the activityArrays in the UserActivities table are the correct (and identical) length,
 * and have the correct start and end dates. Drop any rows that are inconsistent.
 *
 * In practice this should never do anything, but it will stop the data getting badly out of sync if
 * something goes wrong.
 */
async function assertTableIntegrity(dataDb: SqlClient) {
  // Step 1: Find all combinations of startDate and endDate
  const dateCombinations = await dataDb.any(`
    SELECT "startDate", "endDate", COUNT(*) AS count
    FROM "UserActivities"
    GROUP BY "startDate", "endDate"
    ORDER BY count DESC;
  `);

  // Check if there is more than one combination
  if (dateCombinations.length > 1) {
    console.error(`UserActivities table has rows with different start and end dates. Dropping rows to fix this.`);
    console.error('Date combinations and their counts:');
    for (let i = 0; i < dateCombinations.length; i++) {
      const dateCombo = dateCombinations[i];
      console.error(`(${i === 0 ? "correct" : "incorrect"}) startDate: ${dateCombo.startDate}, endDate: ${dateCombo.endDate}, count: ${dateCombo.count}`);
      // get first 10 user ids with this date combo
      const userIds = (await dataDb.any(`
        SELECT "visitorId"
        FROM "UserActivities"
        WHERE "startDate" = $1 AND "endDate" = $2 AND "type" = 'userId'
        LIMIT 10;
      `, [dateCombo.startDate, dateCombo.endDate])).map((row) => row.visitorId);
      // get first 10 client ids with this date combo
      const clientIds = (await dataDb.any(`
        SELECT "visitorId"
        FROM "UserActivities"
        WHERE "startDate" = $1 AND "endDate" = $2 AND "type" = 'clientId'
        LIMIT 10;
      `, [dateCombo.startDate, dateCombo.endDate])).map((row) => row.visitorId);
      console.error(`First 10 userIds: ${userIds.join(', ')}`);
      console.error(`First 10 clientIds: ${clientIds.join(', ')}`);
    }

    // Get the most common combination
    const { startDate: mostCommonStartDate, endDate: mostCommonEndDate } = dateCombinations[0];

    // Step 1.1: Delete rows with different startDate and endDate, keeping the most common combination
    // FIXME currently there is a bug where rows get dropped, add this back in when that is fixed
    // await dataDb.none(`
    //   DELETE FROM "UserActivities"
    //   WHERE "startDate" <> $1 OR "endDate" <> $2;
    // `, [mostCommonStartDate, mostCommonEndDate]);
  }

  // Step 2: Check if the array of activity has the correct length for each row
  const correctActivityLength = await dataDb.one(`
    SELECT EXTRACT(EPOCH FROM (MIN("endDate") - MIN("startDate"))) / 3600 AS correct_length
    FROM "UserActivities";
  `);
  const correctActivityLengthInt = parseInt(correctActivityLength.correct_length);

  // Exit early if the table is empty
  if (!correctActivityLengthInt) return

  // Delete rows with an array of activity that is the wrong length
  const arrayLengths = await dataDb.any(`
  SELECT array_length("activityArray", 1) as length, COUNT(*) AS count
    FROM "UserActivities"
    WHERE array_length("activityArray", 1) <> $1
    GROUP BY array_length("activityArray", 1)
    ORDER BY count DESC;
  `, [correctActivityLengthInt]);

  // Log the incorrect array lengths and their counts
  if (arrayLengths.length > 0) {
    console.error(`UserActivities table has rows with arrays of the incorrect length. Dropping rows to fix this.`);
    console.error('Incorrect array lengths and their counts:');
    for (const arrayLength of arrayLengths) {
      console.error(`length: ${arrayLength.length}, count: ${arrayLength.count}`);
      // get first 10 user ids with this array length
      const userIds = (await dataDb.any(`
        SELECT "visitorId"
        FROM "UserActivities"
        WHERE array_length("activityArray", 1) = $1 AND "type" = 'userId'
        LIMIT 10;
      `, [parseInt(arrayLength.length)])).map((row) => row.visitorId);
      // get first 10 client ids with this array length
      const clientIds = (await dataDb.any(`
        SELECT "visitorId"
        FROM "UserActivities"
        WHERE array_length("activityArray", 1) = $1 AND "type" = 'clientId'
        LIMIT 10;
      `, [parseInt(arrayLength.length)])).map((row) => row.visitorId);
      console.error(`First 10 userIds: ${userIds.join(', ')}`);
      console.error(`First 10 clientIds: ${clientIds.join(', ')}`);
    };
  }

  // FIXME currently there is a bug where rows get dropped, add this back in when that is fixed
  // await dataDb.none(`
  //   DELETE FROM "UserActivities"
  //   WHERE array_length("activityArray", 1) <> $1;
  // `, [correctActivityLengthInt]);
}


/**
 * Get the start and end date for the next user activity update. startDate will be the end date of the
 * current rows in the UserActivities table, or 7 days ago if there are no rows. endDate will be the current time,
 * minus a day so as to ignore the current day's activity.
 */
async function getStartEndDate(dataDb: SqlClient) {
  const { last_end_date, last_start_date } = await dataDb.one(`
    SELECT MAX("endDate") AS last_end_date, MIN("startDate") AS last_start_date
    FROM "UserActivities";
  `);

  const yesterday = new Date(new Date().getTime() - (24 * 60 * 60 * 1000));
  const fallbackStartDate = new Date(yesterday.getTime() - (7 * 24 * 60 * 60 * 1000));
  const lastEndDate = last_end_date ? new Date(last_end_date) : fallbackStartDate;
  const prevStartDate = last_start_date ? new Date(last_start_date) : undefined;
  
  // Round down lastEndDate to the nearest hour
  lastEndDate.setMinutes(0, 0, 0);

  const newEndDate = new Date(yesterday.getTime());
  // Round down endDate to the nearest hour
  newEndDate.setHours(newEndDate.getHours(), 0, 0, 0);

  return {
    prevStartDate,
    updateStartDate: lastEndDate,
    updateEndDate: newEndDate,
  };
}

interface ConcatNewActivityParams {
  dataDb: SqlClient;
  newActivityData: ActivityWindowData[];
  prevStartDate: Date;
  updateStartDate: Date;
  updateEndDate: Date;
  visitorIdType: 'userId' | 'clientId';
  log: (...args: any[]) => void,
}

/**
 * Update UserActivities table with new activity data
 *
 * After this function is run:
 *  - Every user user who was active in last ACTIVITY_WINDOW_HOURS (21 days)
 *    will have an array of activity representing their activity in each hour.
 *    All of these arrays will be the same length (i.e. we zero-pad as necessary)
 *  - Rows from inactive users will be deleted
 */
async function concatNewActivity({
  dataDb,
  newActivityData,
  prevStartDate,
  updateStartDate,
  updateEndDate,
  visitorIdType,
  log,
}: ConcatNewActivityParams) {
  if (updateEndDate.getTime() <= updateStartDate.getTime()) {
    log('No new activity data to update');
    return;
  }
  // validate against SQL injection attacks (userId and clientId are inserted verbatim from the analytics event)
  const cleanedActivityData = newActivityData.filter(({userOrClientId}) => userOrClientId.length <= 17 || !userOrClientId.match(/^[A-Za-z0-9]+$/))

  // Get the existing user IDs from the UserActivities table, required to distinguish between newly active users and existing users
  const existingUserIds = (await dataDb.any(`
    SELECT "visitorId" FROM "UserActivities" where "type" = '${visitorIdType}';
  `)).map(({ visitorId }) => visitorId);
  // Go back to the start of the activity window, up to a maximum of ACTIVITY_WINDOW_HOURS prior to the end time
  // Cast is ok, max function is being cautious about an empty array
  const paddedStartDate = max([prevStartDate, new Date(updateEndDate.getTime() - (ACTIVITY_WINDOW_HOURS * 60 * 60 * 1000))]) as Date;
  
  // -- First case: Add rows for users who are newly active (zero padding the end as necessary)

  // Prepare the new user data to be inserted in the UserActivities table
  const newUsersData = cleanedActivityData
    .filter(({ userOrClientId }) => !existingUserIds.includes(userOrClientId))
    .map(({ userOrClientId, activityArray: activity_array }) => {
      // paddedArray should be [...activity_array], plus zero padding going back to paddedStartDate:
      const zeroPaddingLength = Math.floor((updateEndDate.getTime() - paddedStartDate.getTime()) / (60 * 60 * 1000)) - activity_array.length
      const zeroPadding = Array(zeroPaddingLength).fill(0);
      const paddedArray = [...activity_array, ...zeroPadding];
      return { userOrClientId, paddedArray };
    });

  // Insert data for users we haven't seen before
  if (newUsersData.length > 0) {
    const newUsersDataChunked = chunk(1000, newUsersData)
    log(`Inserting ${newUsersData.length} new rows into UserActivities table, in ${newUsersDataChunked.length} chunks`)
    
    for (const dataChunk of newUsersDataChunked) {
      const placeholders = dataChunk.map((_, index) => `($${(index * 5) + 1}, $${(index * 5) + 2}, $${(index * 5) + 3}, $${(index * 5) + 4}, $${(index * 5) + 5}, '${visitorIdType}')`).join(', ');
      const insertQuery = `
        INSERT INTO "UserActivities" ("_id", "visitorId", "activityArray", "startDate", "endDate", "type")
        VALUES ${placeholders}
      `;
      const queryParams = dataChunk.flatMap(({ userOrClientId, paddedArray }) => {
        return [randomId(), userOrClientId, paddedArray, paddedStartDate.toISOString(), updateEndDate.toISOString()];
      });
      await dataDb.none(insertQuery, queryParams);
    }
  }

  // -- Second case: Append the new activity to users' rows who were previously active.
  // Note the truncation ([:${ACTIVITY_WINDOW_HOURS}]) to ensure that the array of activity is the correct length
  const existingUsersData = cleanedActivityData.filter(({ userOrClientId }) => existingUserIds.includes(userOrClientId));
  if (existingUsersData.length > 0) {
    const existingUsersDataChunked = chunk(1000, existingUsersData)
    log(`Updating ${existingUsersData.length} existing rows for which there is new activity, in ${existingUsersDataChunked.length} chunks`)
    
    for (const dataChunk of existingUsersDataChunked) {
      const tempTableValues = dataChunk.map(({ userOrClientId, activityArray: activity_array }) => `('${userOrClientId}', ARRAY[${activity_array.join(', ')}])`).join(', ');
      const updateQuery = `
        WITH new_activity AS (
          SELECT * FROM (VALUES ${tempTableValues}) AS t(user_id, activity_array)
        )
        UPDATE "UserActivities"
        SET
          "activityArray" = (array_cat(new_activity.activity_array, "UserActivities"."activityArray"))[:${ACTIVITY_WINDOW_HOURS}],
          "startDate" = $1,
          "endDate" = $2
        FROM new_activity
        WHERE
          "UserActivities"."visitorId" = new_activity.user_id
          AND "UserActivities"."type" = '${visitorIdType}';
      `;
      await dataDb.none(updateQuery, [paddedStartDate.toISOString(), updateEndDate.toISOString()]);
    }
  }

  // -- Third case: Zero-pad the start of rows for users who were previously active but have no new activity
  // (i.e. they have existing rows in the table but have not been active since we last updated the table)
  const inactiveExistingUserIds = existingUserIds.filter(id => !existingUsersData.some(({ userOrClientId }) => userOrClientId === id));
  if (inactiveExistingUserIds.length > 0) {
    const newActivityHours = Math.round((updateEndDate.getTime() - updateStartDate.getTime()) / (1000 * 60 * 60));
    const inactiveExistingUserIdsChunked = chunk(1000, inactiveExistingUserIds)
    log(`Adding zero-padding to ${inactiveExistingUserIds.length} existing rows for which there is no new activity, in ${inactiveExistingUserIdsChunked.length} chunks`)

    for (const userIdsChunk of inactiveExistingUserIdsChunked) {
      const updateQuery = `
        UPDATE "UserActivities"
        SET
          "activityArray" = (array_cat(ARRAY[${Array(newActivityHours).fill(0).join(', ')}]::double precision[], "UserActivities"."activityArray"))[:${ACTIVITY_WINDOW_HOURS}],
          "startDate" = $1,
          "endDate" = $2
        WHERE
          "UserActivities"."visitorId" IN (${userIdsChunk.map((_, index) => `$${index + 3}`).join(', ')})
          AND "UserActivities"."type" = '${visitorIdType}';
      `;
      const queryParams = [paddedStartDate.toISOString(), updateEndDate.toISOString(), ...userIdsChunk];
      await dataDb.none(updateQuery, queryParams);
    }
  }

  // -- Fourth case: Every user has had their activity updated now. Remove rows for users who are no longer active in the last
  // ACTIVITY_WINDOW_HOURS (i.e. the activity array is all zeros)
  const countQuery = `
    SELECT COUNT(*) as delete_count FROM "UserActivities"
    WHERE
      "UserActivities"."type" = '${visitorIdType}'
      AND array_position("UserActivities"."activityArray", 1) IS NULL; -- no "1" in the array
  `;
  const deleteQuery = `
    DELETE FROM "UserActivities"
    WHERE
      "UserActivities"."type" = '${visitorIdType}'
      AND array_position("UserActivities"."activityArray", 1) IS NULL; -- no "1" in the array
  `;
  const count = (await dataDb.one(countQuery)).delete_count;
  log(`Deleting ${count} rows for users who have no activity in the last ${ACTIVITY_WINDOW_HOURS} hours`)
  await dataDb.none(deleteQuery);
}

/**
 * Update the UserActivities table with the latest activity data from the analytics database
 *
 * The UserActivities contains an array going back ACTIVITY_WINDOW_HOURS (22 days, starting from 24 hours ago) hours, with each element
 * indicating whether a user was active in that hour. The operation to fetch this from the analytics database is expensive, so this
 * runs every 3 hours to update the table with the latest data.
 *
 * After this function is run:
 *  - Every user user who was active in last ACTIVITY_WINDOW_HOURS (21 days)
 *    will have a complete array representing their activity in each hour.
 *    All of these arrays will be the same length (i.e. we zero-pad as necessary)
 *  - Rows from inactive users will be deleted
 */
export async function updateUserActivities(props?: {
  updateStartDate?: Date,
  updateEndDate?: Date,
  randomWait?: boolean,
  silent?: boolean,
}) {
  const log = props?.silent || isAnyTest ? console.log : () => {};

  const dataDb = getSqlClientOrThrow();

  if (props?.randomWait) {
    // sleep for random amount of time up to 10 seconds
    // FIXME remove once we have a better solution for not running this function twice
    const sleepTime = Math.random() * 10000;
    log(`Sleeping for ${sleepTime.toFixed(0)}ms before updating user activity to avoid multiple jobs running at the same time`)
    await new Promise((resolve) => setTimeout(resolve, sleepTime));
  }

  await assertTableIntegrity(dataDb);
  const { prevStartDate, updateStartDate, updateEndDate } = {...(await getStartEndDate(dataDb)), ...props};

  // Get the most recent activity data from the analytics database
  const newActivityData = await getUserActivityData(updateStartDate, updateEndDate);

  log(`Updating user activity for ${newActivityData.length} users between ${updateStartDate} and ${updateEndDate}`);

  const userActivityData = newActivityData
    .filter(factor => factor.userOrClientId?.startsWith('u:'))
    .map(factor => ({...factor, userOrClientId: factor.userOrClientId.slice(2)}));
  const clientActivityData = newActivityData
    .filter(factor => factor.userOrClientId?.startsWith('c:'))
    .map(factor => ({...factor, userOrClientId: factor.userOrClientId.slice(2)}));

  // Update the UserActivities table with the new activity data
  await concatNewActivity({dataDb, newActivityData: userActivityData, prevStartDate: prevStartDate ?? updateStartDate, updateStartDate, updateEndDate, visitorIdType: 'userId', log});
  await concatNewActivity({dataDb, newActivityData: clientActivityData, prevStartDate: prevStartDate ?? updateStartDate, updateStartDate, updateEndDate, visitorIdType: 'clientId', log});
}

/**
 * Clear all UserActivity data and backfill it all the way back to ACTIVITY_WINDOW_HOURS ago.
 * This takes about 10 minutes on prod (EA Forum)
 */
export async function backfillUserActivities() {
  const dataDb = getSqlClientOrThrow();

  // Clear the current data in the UserActivities collection
  await dataDb.none(`DELETE FROM "UserActivities";`);

  const yesterday = new Date(new Date().getTime() - (24 * 60 * 60 * 1000));
  yesterday.setMinutes(0);
  yesterday.setSeconds(0);
  yesterday.setMilliseconds(0);

  // Calculate the starting date for the backfill (ACTIVITY_WINDOW_HOURS ago)
  const startDate = new Date(yesterday);
  startDate.setHours(startDate.getHours() - ACTIVITY_WINDOW_HOURS);

  // Loop over the range of dates in 24-hour increments
  for (let currentDate = startDate; currentDate < yesterday; currentDate.setHours(currentDate.getHours() + 24)) {
    const endDate = new Date(currentDate);
    endDate.setHours(endDate.getHours() + 24);
    
    // Make sure we don't go past the final end date
    if (endDate > yesterday) {
      endDate.setTime(yesterday.getTime());
    }

    // Update the UserActivities table with the activity data for the current date range
    await updateUserActivities({ updateStartDate: currentDate, updateEndDate: endDate });
  }
}


if (isEAForum || isLW ) {
  addCronJob({
    name: 'updateUserActivitiesCron',
    interval: 'every 3 hours',
    async job() {
      await updateUserActivities({randomWait: true});
    }
  });
}

Vulcan.updateUserActivities = updateUserActivities;
Vulcan.backfillUserActivities = backfillUserActivities;

1	/* eslint-disable no-console */	55✔
2	/* See lib/collections/useractivities/collection.ts for a high-level overview */
3	import chunk from 'lodash/fp/chunk';	18✔
4	import max from 'lodash/fp/max';	18✔
5	import { isEAForum, isLW } from '../../lib/instanceSettings';	18✔
6	import { randomId } from '../../lib/random';	18✔
7	import { getSqlClientOrThrow } from '@/server/sql/sqlClient';	18✔
8	import { addCronJob } from '../cronUtil';	18✔
9	import { Vulcan } from '../vulcan-lib';	18✔
10	import { ActivityWindowData, getUserActivityData } from './getUserActivityData';	18✔
11	import { isAnyTest } from '../../lib/executionEnvironment';	18✔
12
13	const ACTIVITY_WINDOW_HOURS = 21 * 24; // 3 weeks	18✔
14
15	/**
16	* Assert that all the activityArrays in the UserActivities table are the correct (and identical) length,
17	* and have the correct start and end dates. Drop any rows that are inconsistent.
18	*
19	* In practice this should never do anything, but it will stop the data getting badly out of sync if
20	* something goes wrong.
21	*/
22	async function assertTableIntegrity(dataDb: SqlClient) {
23	// Step 1: Find all combinations of startDate and endDate
24	const dateCombinations = await dataDb.any(`	1✔
25	SELECT "startDate", "endDate", COUNT(*) AS count
26	FROM "UserActivities"
27	GROUP BY "startDate", "endDate"
28	ORDER BY count DESC;
29	`);
30
31	// Check if there is more than one combination
32	if (dateCombinations.length > 1) {	1!
UNCOV 33	console.error(`UserActivities table has rows with different start and end dates. Dropping rows to fix this.`);	×
34	console.error('Date combinations and their counts:');	×
35	for (let i = 0; i < dateCombinations.length; i++) {	×
36	const dateCombo = dateCombinations[i];	×
37	console.error(`(${i === 0 ? "correct" : "incorrect"}) startDate: ${dateCombo.startDate}, endDate: ${dateCombo.endDate}, count: ${dateCombo.count}`);	×
38	// get first 10 user ids with this date combo
39	const userIds = (await dataDb.any(`	×
40	SELECT "visitorId"
41	FROM "UserActivities"
42	WHERE "startDate" = $1 AND "endDate" = $2 AND "type" = 'userId'
43	LIMIT 10;
44	`, [dateCombo.startDate, dateCombo.endDate])).map((row) => row.visitorId);	×
45	// get first 10 client ids with this date combo
46	const clientIds = (await dataDb.any(`	×
47	SELECT "visitorId"
48	FROM "UserActivities"
49	WHERE "startDate" = $1 AND "endDate" = $2 AND "type" = 'clientId'
50	LIMIT 10;
51	`, [dateCombo.startDate, dateCombo.endDate])).map((row) => row.visitorId);	×
52	console.error(`First 10 userIds: ${userIds.join(', ')}`);	×
53	console.error(`First 10 clientIds: ${clientIds.join(', ')}`);	×
54	}
55
56	// Get the most common combination
57	const { startDate: mostCommonStartDate, endDate: mostCommonEndDate } = dateCombinations[0];	×
58
59	// Step 1.1: Delete rows with different startDate and endDate, keeping the most common combination
60	// FIXME currently there is a bug where rows get dropped, add this back in when that is fixed
61	// await dataDb.none(`
62	// DELETE FROM "UserActivities"
63	// WHERE "startDate" <> $1 OR "endDate" <> $2;
64	// `, [mostCommonStartDate, mostCommonEndDate]);
65	}
66
67	// Step 2: Check if the array of activity has the correct length for each row
68	const correctActivityLength = await dataDb.one(`	1✔
69	SELECT EXTRACT(EPOCH FROM (MIN("endDate") - MIN("startDate"))) / 3600 AS correct_length
70	FROM "UserActivities";
71	`);
72	const correctActivityLengthInt = parseInt(correctActivityLength.correct_length);	1✔
73
74	// Exit early if the table is empty
75	if (!correctActivityLengthInt) return	1!
76
77	// Delete rows with an array of activity that is the wrong length
78	const arrayLengths = await dataDb.any(`	1✔
79	SELECT array_length("activityArray", 1) as length, COUNT(*) AS count
80	FROM "UserActivities"
81	WHERE array_length("activityArray", 1) <> $1
82	GROUP BY array_length("activityArray", 1)
83	ORDER BY count DESC;
84	`, [correctActivityLengthInt]);
85
86	// Log the incorrect array lengths and their counts
87	if (arrayLengths.length > 0) {	1!
88	console.error(`UserActivities table has rows with arrays of the incorrect length. Dropping rows to fix this.`);	×
89	console.error('Incorrect array lengths and their counts:');	×
90	for (const arrayLength of arrayLengths) {	×
91	console.error(`length: ${arrayLength.length}, count: ${arrayLength.count}`);	×
92	// get first 10 user ids with this array length
93	const userIds = (await dataDb.any(`	×
94	SELECT "visitorId"
95	FROM "UserActivities"
96	WHERE array_length("activityArray", 1) = $1 AND "type" = 'userId'
97	LIMIT 10;
98	`, [parseInt(arrayLength.length)])).map((row) => row.visitorId);	×
99	// get first 10 client ids with this array length
100	const clientIds = (await dataDb.any(`	×
101	SELECT "visitorId"
102	FROM "UserActivities"
103	WHERE array_length("activityArray", 1) = $1 AND "type" = 'clientId'
104	LIMIT 10;
105	`, [parseInt(arrayLength.length)])).map((row) => row.visitorId);	×
106	console.error(`First 10 userIds: ${userIds.join(', ')}`);	×
107	console.error(`First 10 clientIds: ${clientIds.join(', ')}`);	×
108	};
109	}
110
111	// FIXME currently there is a bug where rows get dropped, add this back in when that is fixed
112	// await dataDb.none(`
113	// DELETE FROM "UserActivities"
114	// WHERE array_length("activityArray", 1) <> $1;
115	// `, [correctActivityLengthInt]);
116	}
117
118
119	/**
120	* Get the start and end date for the next user activity update. startDate will be the end date of the
121	* current rows in the UserActivities table, or 7 days ago if there are no rows. endDate will be the current time,
122	* minus a day so as to ignore the current day's activity.
123	*/
124	async function getStartEndDate(dataDb: SqlClient) {
125	const { last_end_date, last_start_date } = await dataDb.one(`	1✔
126	SELECT MAX("endDate") AS last_end_date, MIN("startDate") AS last_start_date
127	FROM "UserActivities";
128	`);
129
130	const yesterday = new Date(new Date().getTime() - (24 * 60 * 60 * 1000));	1✔
131	const fallbackStartDate = new Date(yesterday.getTime() - (7 * 24 * 60 * 60 * 1000));	1✔
132	const lastEndDate = last_end_date ? new Date(last_end_date) : fallbackStartDate;	1!
133	const prevStartDate = last_start_date ? new Date(last_start_date) : undefined;	1!
134
135	// Round down lastEndDate to the nearest hour
136	lastEndDate.setMinutes(0, 0, 0);	1✔
137
138	const newEndDate = new Date(yesterday.getTime());	1✔
139	// Round down endDate to the nearest hour
140	newEndDate.setHours(newEndDate.getHours(), 0, 0, 0);	1✔
141
142	return {	1✔
143	prevStartDate,
144	updateStartDate: lastEndDate,
145	updateEndDate: newEndDate,
146	};
147	}
148
149	interface ConcatNewActivityParams {
150	dataDb: SqlClient;
151	newActivityData: ActivityWindowData[];
152	prevStartDate: Date;
153	updateStartDate: Date;
154	updateEndDate: Date;
155	visitorIdType: 'userId' \| 'clientId';
156	log: (...args: any[]) => void,
157	}
158
159	/**
160	* Update UserActivities table with new activity data
161	*
162	* After this function is run:
163	* - Every user user who was active in last ACTIVITY_WINDOW_HOURS (21 days)
164	* will have an array of activity representing their activity in each hour.
165	* All of these arrays will be the same length (i.e. we zero-pad as necessary)
166	* - Rows from inactive users will be deleted
167	*/
168	async function concatNewActivity({
169	dataDb,
170	newActivityData,
171	prevStartDate,
172	updateStartDate,
173	updateEndDate,
174	visitorIdType,
175	log,
176	}: ConcatNewActivityParams) {
177	if (updateEndDate.getTime() <= updateStartDate.getTime()) {	2!
NEW 178	log('No new activity data to update');	×
179	return;	×
180	}
181	// validate against SQL injection attacks (userId and clientId are inserted verbatim from the analytics event)
182	const cleanedActivityData = newActivityData.filter(({userOrClientId}) => userOrClientId.length <= 17 \|\| !userOrClientId.match(/^[A-Za-z0-9]+$/))	3!
183
184	// Get the existing user IDs from the UserActivities table, required to distinguish between newly active users and existing users
185	const existingUserIds = (await dataDb.any(`	2✔
186	SELECT "visitorId" FROM "UserActivities" where "type" = '${visitorIdType}';
187	`)).map(({ visitorId }) => visitorId);	3✔
188	// Go back to the start of the activity window, up to a maximum of ACTIVITY_WINDOW_HOURS prior to the end time
189	// Cast is ok, max function is being cautious about an empty array
190	const paddedStartDate = max([prevStartDate, new Date(updateEndDate.getTime() - (ACTIVITY_WINDOW_HOURS * 60 * 60 * 1000))]) as Date;	2✔
191
192	// -- First case: Add rows for users who are newly active (zero padding the end as necessary)
193
194	// Prepare the new user data to be inserted in the UserActivities table
195	const newUsersData = cleanedActivityData	2✔
196	.filter(({ userOrClientId }) => !existingUserIds.includes(userOrClientId))	3✔
197	.map(({ userOrClientId, activityArray: activity_array }) => {
198	// paddedArray should be [...activity_array], plus zero padding going back to paddedStartDate:
199	const zeroPaddingLength = Math.floor((updateEndDate.getTime() - paddedStartDate.getTime()) / (60 * 60 * 1000)) - activity_array.length	2✔
200	const zeroPadding = Array(zeroPaddingLength).fill(0);	2✔
201	const paddedArray = [...activity_array, ...zeroPadding];	2✔
202	return { userOrClientId, paddedArray };	2✔
203	});
204
205	// Insert data for users we haven't seen before
206	if (newUsersData.length > 0) {	2!
207	const newUsersDataChunked = chunk(1000, newUsersData)	2✔
208	log(`Inserting ${newUsersData.length} new rows into UserActivities table, in ${newUsersDataChunked.length} chunks`)	2✔
209
210	for (const dataChunk of newUsersDataChunked) {	2✔
211	const placeholders = dataChunk.map((_, index) => `($${(index * 5) + 1}, $${(index * 5) + 2}, $${(index * 5) + 3}, $${(index * 5) + 4}, $${(index * 5) + 5}, '${visitorIdType}')`).join(', ');	2✔
212	const insertQuery = `	2✔
213	INSERT INTO "UserActivities" ("_id", "visitorId", "activityArray", "startDate", "endDate", "type")
214	VALUES ${placeholders}
215	`;
216	const queryParams = dataChunk.flatMap(({ userOrClientId, paddedArray }) => {	2✔
217	return [randomId(), userOrClientId, paddedArray, paddedStartDate.toISOString(), updateEndDate.toISOString()];	2✔
218	});
219	await dataDb.none(insertQuery, queryParams);	2✔
220	}
221	}
222
223	// -- Second case: Append the new activity to users' rows who were previously active.
224	// Note the truncation ([:${ACTIVITY_WINDOW_HOURS}]) to ensure that the array of activity is the correct length
225	const existingUsersData = cleanedActivityData.filter(({ userOrClientId }) => existingUserIds.includes(userOrClientId));	3✔
226	if (existingUsersData.length > 0) {	2✔
227	const existingUsersDataChunked = chunk(1000, existingUsersData)	1✔
228	log(`Updating ${existingUsersData.length} existing rows for which there is new activity, in ${existingUsersDataChunked.length} chunks`)	1✔
229
230	for (const dataChunk of existingUsersDataChunked) {	1✔
231	const tempTableValues = dataChunk.map(({ userOrClientId, activityArray: activity_array }) => `('${userOrClientId}', ARRAY[${activity_array.join(', ')}])`).join(', ');	1✔
232	const updateQuery = `	1✔
233	WITH new_activity AS (
234	SELECT * FROM (VALUES ${tempTableValues}) AS t(user_id, activity_array)
235	)
236	UPDATE "UserActivities"
237	SET
238	"activityArray" = (array_cat(new_activity.activity_array, "UserActivities"."activityArray"))[:${ACTIVITY_WINDOW_HOURS}],
239	"startDate" = $1,
240	"endDate" = $2
241	FROM new_activity
242	WHERE
243	"UserActivities"."visitorId" = new_activity.user_id
244	AND "UserActivities"."type" = '${visitorIdType}';
245	`;
246	await dataDb.none(updateQuery, [paddedStartDate.toISOString(), updateEndDate.toISOString()]);	1✔
247	}
248	}
249
250	// -- Third case: Zero-pad the start of rows for users who were previously active but have no new activity
251	// (i.e. they have existing rows in the table but have not been active since we last updated the table)
252	const inactiveExistingUserIds = existingUserIds.filter(id => !existingUsersData.some(({ userOrClientId }) => userOrClientId === id));	3✔
253	if (inactiveExistingUserIds.length > 0) {	2!
254	const newActivityHours = Math.round((updateEndDate.getTime() - updateStartDate.getTime()) / (1000 * 60 * 60));	2✔
255	const inactiveExistingUserIdsChunked = chunk(1000, inactiveExistingUserIds)	2✔
256	log(`Adding zero-padding to ${inactiveExistingUserIds.length} existing rows for which there is no new activity, in ${inactiveExistingUserIdsChunked.length} chunks`)	2✔
257
258	for (const userIdsChunk of inactiveExistingUserIdsChunked) {	2✔
259	const updateQuery = `	2✔
260	UPDATE "UserActivities"
261	SET
262	"activityArray" = (array_cat(ARRAY[${Array(newActivityHours).fill(0).join(', ')}]::double precision[], "UserActivities"."activityArray"))[:${ACTIVITY_WINDOW_HOURS}],
263	"startDate" = $1,
264	"endDate" = $2
265	WHERE
266	"UserActivities"."visitorId" IN (${userIdsChunk.map((_, index) => `$${index + 3}`).join(', ')})	2✔
267	AND "UserActivities"."type" = '${visitorIdType}';
268	`;
269	const queryParams = [paddedStartDate.toISOString(), updateEndDate.toISOString(), ...userIdsChunk];	2✔
270	await dataDb.none(updateQuery, queryParams);	2✔
271	}
272	}
273
274	// -- Fourth case: Every user has had their activity updated now. Remove rows for users who are no longer active in the last
275	// ACTIVITY_WINDOW_HOURS (i.e. the activity array is all zeros)
276	const countQuery = `	2✔
277	SELECT COUNT(*) as delete_count FROM "UserActivities"
278	WHERE
279	"UserActivities"."type" = '${visitorIdType}'
280	AND array_position("UserActivities"."activityArray", 1) IS NULL; -- no "1" in the array
281	`;
282	const deleteQuery = `	2✔
283	DELETE FROM "UserActivities"
284	WHERE
285	"UserActivities"."type" = '${visitorIdType}'
286	AND array_position("UserActivities"."activityArray", 1) IS NULL; -- no "1" in the array
287	`;
288	const count = (await dataDb.one(countQuery)).delete_count;	2✔
289	log(`Deleting ${count} rows for users who have no activity in the last ${ACTIVITY_WINDOW_HOURS} hours`)	2✔
290	await dataDb.none(deleteQuery);	2✔
291	}
292
293	/**
294	* Update the UserActivities table with the latest activity data from the analytics database
295	*
296	* The UserActivities contains an array going back ACTIVITY_WINDOW_HOURS (22 days, starting from 24 hours ago) hours, with each element
297	* indicating whether a user was active in that hour. The operation to fetch this from the analytics database is expensive, so this
298	* runs every 3 hours to update the table with the latest data.
299	*
300	* After this function is run:
301	* - Every user user who was active in last ACTIVITY_WINDOW_HOURS (21 days)
302	* will have a complete array representing their activity in each hour.
303	* All of these arrays will be the same length (i.e. we zero-pad as necessary)
304	* - Rows from inactive users will be deleted
305	*/
306	export async function updateUserActivities(props?: {
307	updateStartDate?: Date,
308	updateEndDate?: Date,
309	randomWait?: boolean,
310	silent?: boolean,
311	}) {
312	const log = props?.silent \|\| isAnyTest ? console.log : () => {};	1!
313
314	const dataDb = getSqlClientOrThrow();	1✔
315
316	if (props?.randomWait) {	1!
317	// sleep for random amount of time up to 10 seconds
318	// FIXME remove once we have a better solution for not running this function twice
319	const sleepTime = Math.random() * 10000;	×
NEW 320	log(`Sleeping for ${sleepTime.toFixed(0)}ms before updating user activity to avoid multiple jobs running at the same time`)	×
321	await new Promise((resolve) => setTimeout(resolve, sleepTime));	×
322	}
323
324	await assertTableIntegrity(dataDb);	1✔
325	const { prevStartDate, updateStartDate, updateEndDate } = {...(await getStartEndDate(dataDb)), ...props};	1✔
326
327	// Get the most recent activity data from the analytics database
328	const newActivityData = await getUserActivityData(updateStartDate, updateEndDate);	1✔
329
330	log(`Updating user activity for ${newActivityData.length} users between ${updateStartDate} and ${updateEndDate}`);	1✔
331
332	const userActivityData = newActivityData	1✔
333	.filter(factor => factor.userOrClientId?.startsWith('u:'))	3!
334	.map(factor => ({...factor, userOrClientId: factor.userOrClientId.slice(2)}));	2✔
335	const clientActivityData = newActivityData	1✔
336	.filter(factor => factor.userOrClientId?.startsWith('c:'))	3!
337	.map(factor => ({...factor, userOrClientId: factor.userOrClientId.slice(2)}));	1✔
338
339	// Update the UserActivities table with the new activity data
340	await concatNewActivity({dataDb, newActivityData: userActivityData, prevStartDate: prevStartDate ?? updateStartDate, updateStartDate, updateEndDate, visitorIdType: 'userId', log});	1!
341	await concatNewActivity({dataDb, newActivityData: clientActivityData, prevStartDate: prevStartDate ?? updateStartDate, updateStartDate, updateEndDate, visitorIdType: 'clientId', log});	1!
342	}
343
344	/**
345	* Clear all UserActivity data and backfill it all the way back to ACTIVITY_WINDOW_HOURS ago.
346	* This takes about 10 minutes on prod (EA Forum)
347	*/
348	export async function backfillUserActivities() {
349	const dataDb = getSqlClientOrThrow();	×
350
351	// Clear the current data in the UserActivities collection
352	await dataDb.none(`DELETE FROM "UserActivities";`);	×
353
354	const yesterday = new Date(new Date().getTime() - (24 * 60 * 60 * 1000));	×
355	yesterday.setMinutes(0);	×
356	yesterday.setSeconds(0);	×
357	yesterday.setMilliseconds(0);	×
358
359	// Calculate the starting date for the backfill (ACTIVITY_WINDOW_HOURS ago)
360	const startDate = new Date(yesterday);	×
361	startDate.setHours(startDate.getHours() - ACTIVITY_WINDOW_HOURS);	×
362
363	// Loop over the range of dates in 24-hour increments
364	for (let currentDate = startDate; currentDate < yesterday; currentDate.setHours(currentDate.getHours() + 24)) {	×
365	const endDate = new Date(currentDate);	×
366	endDate.setHours(endDate.getHours() + 24);	×
367
368	// Make sure we don't go past the final end date
369	if (endDate > yesterday) {	×
370	endDate.setTime(yesterday.getTime());	×
371	}
372
373	// Update the UserActivities table with the activity data for the current date range
374	await updateUserActivities({ updateStartDate: currentDate, updateEndDate: endDate });	×
375	}
376	}
377
378
379	if (isEAForum \|\| isLW ) {	18!
380	addCronJob({	18✔
381	name: 'updateUserActivitiesCron',
382	interval: 'every 3 hours',
383	async job() {
384	await updateUserActivities({randomWait: true});	×
385	}
386	});
387	}
388
389	Vulcan.updateUserActivities = updateUserActivities;	18✔
390	Vulcan.backfillUserActivities = backfillUserActivities;	18✔

ForumMagnum / ForumMagnum / 10411869195

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous