8730738677

Committed 18 Apr 2024 12:55AM UTC coverage: 32.002% (+0.06%) from 31.945%

Build # 8730738677

Build Type

push

github

Committed by

b0b3rt

Commit Message

switch to third app

Run Details

4630 of 37776 branches covered (12.26%)

Branch coverage included in aggregate %.

2 of 3 new or added lines in 1 file covered. (66.67%)

1 existing line in 1 file now uncovered.

27049 of 61215 relevant lines covered (44.19%)

182.91 hits per line

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

11.49

/packages/lesswrong/server/scripts/backfillBigquery.ts

import { readFile, writeFile } from "fs/promises";
import { htmlToTextDefault } from "../../lib/htmlToText";
import { getSqlClientOrThrow } from "../../lib/sql/sqlClient";
import { filterNonnull } from "../../lib/utils/typeGuardUtils";
import { Globals, createAdminContext } from "../vulcan-lib";
import findByIds from "../vulcan-lib/findbyids";
import { postGetPageUrl } from "../../lib/collections/posts/helpers";
import { DocumentServiceClient, UserEventServiceClient, protos } from "@google-cloud/discoveryengine";
import groupBy from "lodash/groupBy";
import ReadStatuses from "../../lib/collections/readStatus/collection";
import chunk from "lodash/chunk";

const GOOGLE_PARENT_DOCUMENTS_PATH = 'projects/lesswrong-recommendations/locations/global/dataStores/datastore-lw-recommendations3_1713386634021/branches/default_branch';
const GOOGLE_PARENT_EVENTS_PATH = 'projects/lesswrong-recommendations/locations/global/dataStores/datastore-lw-recommendations3_1713386634021';

export const getGoogleDocumentServiceClientOrThrow = (() => {
  let client: DocumentServiceClient;

  return () => {
    if (!client) {
      client = new DocumentServiceClient();
    }

    return client;
  };
})();


export const getGoogleUserEventServiceClientOrThrow = (() => {
  let client: UserEventServiceClient;

  return () => {
    if (!client) {
      client = new UserEventServiceClient();
    }

    return client;
  };
})();

function getNextOffsetDate<T extends HasCreatedAtType>(currentOffsetDate: Date, batch: T[]) {
  const nextOffsetDate = batch.slice(-1)[0].createdAt;
  if (currentOffsetDate.getTime() === nextOffsetDate.getTime()) {
    // eslint-disable-next-line no-console
    console.log(`Next batch offset date is the same as previous offset date: ${currentOffsetDate.toISOString()}.  If this seems like an early return, investigate!`);
    return;
  }

  // eslint-disable-next-line no-console
  console.log(`Next batch offset date: ${nextOffsetDate.toISOString()}`);

  return nextOffsetDate;
}

const postBatchQuery = `
  WITH selected_posts AS (
    SELECT
      p._id,
      p."userId",
      p."coauthorStatuses"
    FROM "Posts" AS p
    WHERE p.draft IS NOT TRUE
      AND p."isFuture" IS FALSE
      AND p."unlisted" IS NOT TRUE
      AND p.rejected IS NOT TRUE
      AND p."hiddenRelatedQuestion" IS NOT TRUE
      AND p."groupId" IS NULL
      AND p.shortform IS NOT TRUE
      AND ((p."authorIsUnreviewed" IS NOT TRUE) OR (p."reviewedByUserId" IS NOT NULL))
      AND p.status = 2
      AND p."createdAt" >= $1
    ORDER BY p."createdAt" ASC
    LIMIT $2
  ), authorships AS ((
    SELECT
      _id,
      "userId"
    FROM
    selected_posts
  )
  UNION ALL (
    SELECT
      _id,
      unnest("coauthorStatuses") ->> 'userId' AS "userId"
    FROM selected_posts
  ))
  SELECT
    p.*,
    CASE WHEN COUNT(t.*) = 0
      THEN '{}'::JSONB[]
      ELSE ARRAY_AGG(JSONB_BUILD_ARRAY(t._id, t.name, t.core))
    END AS tags,
    (SELECT ARRAY_AGG(COALESCE(u."displayName", u."username")) FROM "Users" AS u WHERE u._id IN (SELECT "userId" FROM authorships WHERE authorships._id = p._id)) AS authors,
    ARRAY_AGG(DISTINCT authorships."userId") AS "authorIds",
    (SELECT COUNT(*) FROM "Votes" AS v WHERE v."documentId" = p._id AND v.cancelled IS NOT TRUE AND v.power > 0 AND v."userId" NOT IN (SELECT "userId" FROM authorships WHERE authorships._id = p._id)) AS "upvoteCount"
  FROM authorships
  INNER JOIN "Posts" AS p
  USING(_id)
  LEFT JOIN "TagRels" AS tr
  ON authorships._id = tr."postId"
  LEFT JOIN "Tags" AS t
  ON tr."tagId" = t._id
  GROUP BY p._id
  ORDER BY p."createdAt" ASC
`;

function getPostBatch(offsetDate: Date) {
  const db = getSqlClientOrThrow();
  const limit = 100;
  return db.any<DbPost & { tags: [tagId: string, tagName: string, core: boolean][], authors: string[], authorIds: string[], upvoteCount: number }>(postBatchQuery, [offsetDate, limit]);
}

interface CreateBigQueryPostRecordArgs {
  post: DbPost;
  context: ResolverContext;
  tags?: {
    _id: string;
    name: string;
    core: boolean;
  }[];
}

async function createBigQueryPostRecord({ post, context, tags }: CreateBigQueryPostRecordArgs) {
  const { Tags } = context;

  const tagIds = Object.entries(post.tagRelevance ?? {}).filter(([_, relevance]: [string, number]) => relevance > 0).map(([tagId]) => tagId)
  tags ??= filterNonnull(await findByIds(Tags, tagIds))
  const tagNames = filterNonnull(tags.map(tag => tag.name))
  const coreTagNames = filterNonnull(tags.filter(tag => tag.core).map(tag => tag.name))

  const postText = htmlToTextDefault(post.contents?.html)

  return {
    title: post.title,
    author: post.author,
    authorId: post.userId,
    karma: post.baseScore,
    body: postText,
    postedAt: post.postedAt,
    tags: tagNames,
    coreTags: coreTagNames,
    curated: !!post.curatedDate,
    frontpage: !!post.frontpageDate,
    draft: !!post.draft,
    lastCommentedAt: post.lastCommentedAt,
  };
}

interface CreateAEPostRecordArgs {
  post: DbPost;
  context: ResolverContext;
  tags?: {
    _id: string;
    name: string;
    core: boolean;
  }[];
  authors?: string[];
  upvoteCount?: number;
}

async function createAEPostRecord({ post, context, tags, authors, upvoteCount }: CreateAEPostRecordArgs) {
  const { Tags } = context;

  const tagIds = Object.entries(post.tagRelevance ?? {}).filter(([_, relevance]: [string, number]) => relevance > 0).map(([tagId]) => tagId);
  tags ??= filterNonnull(await findByIds(Tags, tagIds));
  const tagNames = filterNonnull(tags.map(tag => tag.name));
  const postText = htmlToTextDefault(post.contents?.html);

  return {
    title: post.title,
    authors,
    score: post.score,
    karma: post.baseScore,
    body: postText,
    postedAt: post.postedAt,
    tags: tagNames,
    commentCount: post.commentCount,
    upvoteCount
  };
}

type GoogleMediaPersonOrgRole = 'director' | 'actor' | 'player' | 'team' | 'league' | 'editor' | 'author' | 'character' | 'contributor' | 'creator' | 'editor' | 'funder' | 'producer' | 'provider' | 'publisher' | 'sponsor' | 'translator' | 'music-by' | 'channel' | 'custom-role';

interface GoogleMediaDocumentMetadata {
  title: string;
  categories: string[];
  uri: string;
  description?: string;
  /**
   * For document recommendation, this field is ignored and the text language is detected automatically.
   * The document can include text in different languages, but duplicating documents to provide text in multiple languages can result in degraded performance.
   */
  language_code?: string;
  images?: Array<{
    uri?: string;
    name?: string;
  }>;
  duration?: string;
  /**
   * The time that the content is available to the end-users. This field identifies the freshness of a content for end-users. The timestamp should conform to RFC 3339 standard.
   */
  available_time: string;
  expire_time?: string;
  media_type?: 'episode' | 'movie' | 'concert' | 'event' | 'live-event' | 'broadcast' | 'tv-series' | 'video-game' | 'clip' | 'vlog' | 'audio' | 'audio-book' | 'music' | 'album' | 'articles' | 'news' | 'radio' | 'podcast' | 'book' | 'sports-game';
  in_languages?: string[];
  country_of_origin?: string;
  filter_tags?: string[];
  hash_tags?: string[];
  content_rating?: string[];
  persons?: Array<{
    name: string;
    role: GoogleMediaPersonOrgRole;
    custom_role?: string;
    rank?: number;
    uri?: string;
  }>;
  organizations?: Array<{
    name: string;
    role: GoogleMediaPersonOrgRole;
    custom_role?: string;
    /**
     * Is this really a string?  That's what their JSON schema says, but it's an int for `persons`...
     */
    rank?: string;
    uri?: string;
  }>;
  aggregate_ratings?: Array<{
    rating_source: string;
    rating_score?: number;
    rating_count?: number;
  }>
}

interface CreateGoogleMediaDocumentMetadataArgs {
  post: DbPost;
  tags?: {
    _id: string;
    name: string;
    core: boolean;
  }[];
  authorIds?: string[];
}

function createGoogleMediaDocumentJson({ post, tags, authorIds }: CreateGoogleMediaDocumentMetadataArgs): protos.google.cloud.discoveryengine.v1.IDocument {
  const tagIds = tags?.map(tag => tag._id) ?? [];
  if (tagIds.length === 0) {
    tagIds.push('none')
  }
  const persons = authorIds?.map(authorId => ({ name: authorId, role: 'author' } as const));

  const metadata: GoogleMediaDocumentMetadata = {
    title: post.title,
    uri: `https://www.lesswrong.com${postGetPageUrl(post)}`,
    available_time: post.postedAt.toISOString(),
    categories: tagIds,
    filter_tags: tagIds,
    persons,
    media_type: 'articles',
  };

  return {
    id: post._id,
    schemaId: 'default_schema',
    jsonData: JSON.stringify(metadata),
  };
}

interface InViewEvent {
  userId: string;
  postId: string;
  /**
   * When generating the table with these events, we forgot to alias the MIN(timestamp) back to `timestamp`.
   */
  min: string;
}

function createGoogleViewItemEvent(eventType: 'view-item'|'media-play', readStatus: DbReadStatus): protos.google.cloud.discoveryengine.v1.IUserEvent {
  const { userId, postId, lastUpdated: timestamp } = readStatus;

  return {
    eventType,
    // TODO - this should be clientId if doing real stuff with it
    userPseudoId: userId, 
    eventTime: {
      seconds: timestamp.getTime() / 1000,
      nanos: 0
    },
    userInfo: { userId },
    documents: [ { id: postId } ],
  };
}

function createGoogleMediaCompleteEvent(inViewEvent: InViewEvent): protos.google.cloud.discoveryengine.v1.IUserEvent {
  const { userId, postId, min } = inViewEvent;

  return {
    eventType: 'media-complete',
    // TODO - this should be clientId if doing real stuff with it
    userPseudoId: userId, 
    eventTime: {
      seconds: new Date(min).getTime() / 1000,
      nanos: 0
    },
    userInfo: { userId },
    documents: [ { id: postId } ],
    mediaInfo: {
      mediaProgressPercentage: 1
    }
  };
}

function createDelimitedJsonString<T>(records: T[]) {
  return records.map(record => JSON.stringify(record)).join('\n');
}

function indexInViewEvents(inViewEvents: InViewEvent[]): Record<string, Record<string, string>> {
  const indexedInViewEvents: Record<string, Record<string, string>> = {};

  // eslint-disable-next-line no-console
  console.log(`Indexing ${inViewEvents.length} inViewEvents`);

  inViewEvents.reduce((acc, { postId, userId, min }) => {
    acc[postId] ??= {};
    acc[postId][userId] = min;
    return acc;
  }, indexedInViewEvents);

  // eslint-disable-next-line no-console
  console.log(`Indexed inViewEvents for ${Object.keys(indexedInViewEvents).length} posts`);

  return indexedInViewEvents;
}

function createViewHomePageEvent({userId, timestamp}: { userId: string, timestamp: string }) {
  return {
    eventType: 'view-home-page',
    userPseudoId: userId,
    eventTime: {
      seconds: new Date(timestamp).getTime() / 1000,
      nanos: 0
    },
    userInfo: { userId }
  };
}

async function backfillPosts(offsetDate?: Date) {
  const adminContext = createAdminContext();
  const db = getSqlClientOrThrow();
  const documentClient = getGoogleDocumentServiceClientOrThrow();
  const userEventClient = getGoogleUserEventServiceClientOrThrow();

  const inViewEvents: InViewEvent[] = JSON.parse((await readFile('/Users/robert/Documents/repos/ForumMagnum/in_view_events_for_google_20240415.json')).toString());
  const indexedInViewEvents = indexInViewEvents(inViewEvents);

  if (!offsetDate) {
    ({ offsetDate } = await db.one<{ offsetDate: Date }>('SELECT MIN("createdAt") AS "offsetDate" FROM "Posts"'));
  }

  // eslint-disable-next-line no-console
  console.log(`Initial post batch offset date: ${offsetDate.toISOString()}`);

  let batch = await getPostBatch(offsetDate);

  try {
    while (batch.length) {
      // eslint-disable-next-line no-console
      console.log(`Post batch size: ${batch.length}.`);

      const postsWithTags = batch.map(({ tags, authors, authorIds, upvoteCount, ...post }) => ({
        post,
        tags: filterNonnull(tags.map(([_id, name, core]) => ({ _id, name, core }))),
        authors,
        authorIds,
        upvoteCount
      }));

      const googleMediaDocuments = postsWithTags.map(createGoogleMediaDocumentJson);
      if (googleMediaDocuments.length) {
        const postIds = postsWithTags.map(({ post }) => post._id);
        const readStatusOperation = () => ReadStatuses.find(
          { postId: { $in: postIds }, isRead: true }, 
          undefined, 
          { _id: 1, userId: 1, postId: 1, lastUpdated: 1 }
        ).fetch();
        
        const [importDocumentsOperation] = await documentClient.importDocuments({ inlineSource: { documents: googleMediaDocuments }, parent: GOOGLE_PARENT_DOCUMENTS_PATH });
        const [[importDocumentsResponse], readStatuses] = await Promise.all([
          importDocumentsOperation.promise(),
          readStatusOperation()
        ]);

        if (importDocumentsResponse.errorSamples?.length) {
          // eslint-disable-next-line no-console
          console.log('Error importing documents', { error: importDocumentsResponse.errorSamples[0] });
        }

        const postReadStatusMap = groupBy(readStatuses, 'postId');
        
        const viewItemEvents = readStatuses.map(readStatus => createGoogleViewItemEvent('view-item', readStatus));
        const mediaPlayEvents = readStatuses.map(readStatus => createGoogleViewItemEvent('media-play', readStatus));

        const inViewEvents = postsWithTags.map(({ post }) => {
          const postReadStatuses = postReadStatusMap[post._id] ?? [];
          const correspondingInViewEvents = postReadStatuses.filter(({ postId, userId }) => !!indexedInViewEvents[postId!]?.[userId]);
          return correspondingInViewEvents.map(({ postId, userId }) => {
            const min = indexedInViewEvents[postId!][userId];
            return { postId, userId, min };
          })
        }).flat();

        const mediaCompleteEvents = inViewEvents.map(createGoogleMediaCompleteEvent);

        const userEvents = [...viewItemEvents, ...mediaPlayEvents, ...mediaCompleteEvents];

        const chunkedUserEvents = chunk(userEvents, 90000);
        for (const chunk of chunkedUserEvents) {
          const [importUserEventsOperation] = await userEventClient.importUserEvents({ inlineSource: { userEvents: chunk }, parent: GOOGLE_PARENT_EVENTS_PATH });
          const [importUserEventsResponse] = await importUserEventsOperation.promise();
          if (importUserEventsResponse.errorSamples?.length) {
            // eslint-disable-next-line no-console
            console.log('Error importing user events', { error: importUserEventsResponse.errorSamples[0] });
          }
        }
      }

      const nextOffsetDate: Date | undefined = getNextOffsetDate(offsetDate, batch);
      if (!nextOffsetDate) {
        return;
      }
      offsetDate = nextOffsetDate;
      batch = await getPostBatch(offsetDate);
    }  
  } catch (err) {
    // eslint-disable-next-line no-console
    console.log(`Error when backfilling BigQuery/Google with posts.  Last offset date: ${offsetDate.toISOString()}`, { err });
  }
}

interface FrontpageView {
  userId: string;
  timestamp: string;
}

async function backfillFrontpageViews(offsetDate?: Date) {
  const userEventClient = getGoogleUserEventServiceClientOrThrow();

  const frontpageViewEvensts: FrontpageView[] = JSON.parse((await readFile('/Users/rbloom/git/lesswrongSuite/logged_in_user_frontpage_loads_20240416.json')).toString());

  const chunkedFrontpageViews = chunk(frontpageViewEvensts, 90000);

  for (const chunk of chunkedFrontpageViews) {
    // eslint-disable-next-line no-console
    console.log(`Processing chunk of ${chunk.length} frontpage views`);
    const userEvents = chunk.map(({ userId, timestamp }) => createViewHomePageEvent({ userId, timestamp }));
    const [importUserEventsOperation] = await userEventClient.importUserEvents({ inlineSource: { userEvents }, parent: GOOGLE_PARENT_EVENTS_PATH });
    const [importUserEventsResponse] = await importUserEventsOperation.promise();
    if (importUserEventsResponse.errorSamples?.length) {
      // eslint-disable-next-line no-console
      console.log('Error importing frontpage loads', { error: importUserEventsResponse.errorSamples[0] });
    }
  }
}


Globals.backfillBigQueryPosts = backfillPosts;
Globals.backfillFrontpageViews = backfillFrontpageViews;

1	import { readFile, writeFile } from "fs/promises";	54✔
2	import { htmlToTextDefault } from "../../lib/htmlToText";	18✔
3	import { getSqlClientOrThrow } from "../../lib/sql/sqlClient";	18✔
4	import { filterNonnull } from "../../lib/utils/typeGuardUtils";	18✔
5	import { Globals, createAdminContext } from "../vulcan-lib";	18✔
6	import findByIds from "../vulcan-lib/findbyids";	18✔
7	import { postGetPageUrl } from "../../lib/collections/posts/helpers";	18✔
8	import { DocumentServiceClient, UserEventServiceClient, protos } from "@google-cloud/discoveryengine";	18✔
9	import groupBy from "lodash/groupBy";	18✔
10	import ReadStatuses from "../../lib/collections/readStatus/collection";	18✔
11	import chunk from "lodash/chunk";	18✔
12
13	const GOOGLE_PARENT_DOCUMENTS_PATH = 'projects/lesswrong-recommendations/locations/global/dataStores/datastore-lw-recommendations3_1713386634021/branches/default_branch';	18✔
14	const GOOGLE_PARENT_EVENTS_PATH = 'projects/lesswrong-recommendations/locations/global/dataStores/datastore-lw-recommendations3_1713386634021';	18✔
15
16	export const getGoogleDocumentServiceClientOrThrow = (() => {	18✔
17	let client: DocumentServiceClient;
18
19	return () => {	18✔
20	if (!client) {	×
21	client = new DocumentServiceClient();	×
22	}
23
24	return client;	×
25	};
26	})();
27
28
29	export const getGoogleUserEventServiceClientOrThrow = (() => {	18✔
30	let client: UserEventServiceClient;
31
32	return () => {	18✔
33	if (!client) {	×
34	client = new UserEventServiceClient();	×
35	}
36
37	return client;	×
38	};
39	})();
40
41	function getNextOffsetDate<T extends HasCreatedAtType>(currentOffsetDate: Date, batch: T[]) {
42	const nextOffsetDate = batch.slice(-1)[0].createdAt;	×
43	if (currentOffsetDate.getTime() === nextOffsetDate.getTime()) {	×
44	// eslint-disable-next-line no-console
45	console.log(`Next batch offset date is the same as previous offset date: ${currentOffsetDate.toISOString()}. If this seems like an early return, investigate!`);	×
46	return;	×
47	}
48
49	// eslint-disable-next-line no-console
50	console.log(`Next batch offset date: ${nextOffsetDate.toISOString()}`);	×
51
52	return nextOffsetDate;	×
53	}
54
55	const postBatchQuery = `	18✔
56	WITH selected_posts AS (
57	SELECT
58	p._id,
59	p."userId",
60	p."coauthorStatuses"
61	FROM "Posts" AS p
62	WHERE p.draft IS NOT TRUE
63	AND p."isFuture" IS FALSE
64	AND p."unlisted" IS NOT TRUE
65	AND p.rejected IS NOT TRUE
66	AND p."hiddenRelatedQuestion" IS NOT TRUE
67	AND p."groupId" IS NULL
68	AND p.shortform IS NOT TRUE
69	AND ((p."authorIsUnreviewed" IS NOT TRUE) OR (p."reviewedByUserId" IS NOT NULL))
70	AND p.status = 2
71	AND p."createdAt" >= $1
72	ORDER BY p."createdAt" ASC
73	LIMIT $2
74	), authorships AS ((
75	SELECT
76	_id,
77	"userId"
78	FROM
79	selected_posts
80	)
81	UNION ALL (
82	SELECT
83	_id,
84	unnest("coauthorStatuses") ->> 'userId' AS "userId"
85	FROM selected_posts
86	))
87	SELECT
88	p.*,
89	CASE WHEN COUNT(t.*) = 0
90	THEN '{}'::JSONB[]
91	ELSE ARRAY_AGG(JSONB_BUILD_ARRAY(t._id, t.name, t.core))
92	END AS tags,
93	(SELECT ARRAY_AGG(COALESCE(u."displayName", u."username")) FROM "Users" AS u WHERE u._id IN (SELECT "userId" FROM authorships WHERE authorships._id = p._id)) AS authors,
94	ARRAY_AGG(DISTINCT authorships."userId") AS "authorIds",
95	(SELECT COUNT(*) FROM "Votes" AS v WHERE v."documentId" = p._id AND v.cancelled IS NOT TRUE AND v.power > 0 AND v."userId" NOT IN (SELECT "userId" FROM authorships WHERE authorships._id = p._id)) AS "upvoteCount"
96	FROM authorships
97	INNER JOIN "Posts" AS p
98	USING(_id)
99	LEFT JOIN "TagRels" AS tr
100	ON authorships._id = tr."postId"
101	LEFT JOIN "Tags" AS t
102	ON tr."tagId" = t._id
103	GROUP BY p._id
104	ORDER BY p."createdAt" ASC
105	`;
106
107	function getPostBatch(offsetDate: Date) {
108	const db = getSqlClientOrThrow();	×
109	const limit = 100;	×
110	return db.any<DbPost & { tags: [tagId: string, tagName: string, core: boolean][], authors: string[], authorIds: string[], upvoteCount: number }>(postBatchQuery, [offsetDate, limit]);	×
111	}
112
113	interface CreateBigQueryPostRecordArgs {
114	post: DbPost;
115	context: ResolverContext;
116	tags?: {
117	_id: string;
118	name: string;
119	core: boolean;
120	}[];
121	}
122
123	async function createBigQueryPostRecord({ post, context, tags }: CreateBigQueryPostRecordArgs) {
124	const { Tags } = context;	×
125
126	const tagIds = Object.entries(post.tagRelevance ?? {}).filter(([_, relevance]: [string, number]) => relevance > 0).map(([tagId]) => tagId)	×
127	tags ??= filterNonnull(await findByIds(Tags, tagIds))	×
128	const tagNames = filterNonnull(tags.map(tag => tag.name))	×
129	const coreTagNames = filterNonnull(tags.filter(tag => tag.core).map(tag => tag.name))	×
130
131	const postText = htmlToTextDefault(post.contents?.html)	×
132
133	return {	×
134	title: post.title,
135	author: post.author,
136	authorId: post.userId,
137	karma: post.baseScore,
138	body: postText,
139	postedAt: post.postedAt,
140	tags: tagNames,
141	coreTags: coreTagNames,
142	curated: !!post.curatedDate,
143	frontpage: !!post.frontpageDate,
144	draft: !!post.draft,
145	lastCommentedAt: post.lastCommentedAt,
146	};
147	}
148
149	interface CreateAEPostRecordArgs {
150	post: DbPost;
151	context: ResolverContext;
152	tags?: {
153	_id: string;
154	name: string;
155	core: boolean;
156	}[];
157	authors?: string[];
158	upvoteCount?: number;
159	}
160
161	async function createAEPostRecord({ post, context, tags, authors, upvoteCount }: CreateAEPostRecordArgs) {
162	const { Tags } = context;	×
163
164	const tagIds = Object.entries(post.tagRelevance ?? {}).filter(([_, relevance]: [string, number]) => relevance > 0).map(([tagId]) => tagId);	×
165	tags ??= filterNonnull(await findByIds(Tags, tagIds));	×
166	const tagNames = filterNonnull(tags.map(tag => tag.name));	×
167	const postText = htmlToTextDefault(post.contents?.html);	×
168
169	return {	×
170	title: post.title,
171	authors,
172	score: post.score,
173	karma: post.baseScore,
174	body: postText,
175	postedAt: post.postedAt,
176	tags: tagNames,
177	commentCount: post.commentCount,
178	upvoteCount
179	};
180	}
181
182	type GoogleMediaPersonOrgRole = 'director' \| 'actor' \| 'player' \| 'team' \| 'league' \| 'editor' \| 'author' \| 'character' \| 'contributor' \| 'creator' \| 'editor' \| 'funder' \| 'producer' \| 'provider' \| 'publisher' \| 'sponsor' \| 'translator' \| 'music-by' \| 'channel' \| 'custom-role';
183
184	interface GoogleMediaDocumentMetadata {
185	title: string;
186	categories: string[];
187	uri: string;
188	description?: string;
189	/**
190	* For document recommendation, this field is ignored and the text language is detected automatically.
191	* The document can include text in different languages, but duplicating documents to provide text in multiple languages can result in degraded performance.
192	*/
193	language_code?: string;
194	images?: Array<{
195	uri?: string;
196	name?: string;
197	}>;
198	duration?: string;
199	/**
200	* The time that the content is available to the end-users. This field identifies the freshness of a content for end-users. The timestamp should conform to RFC 3339 standard.
201	*/
202	available_time: string;
203	expire_time?: string;
204	media_type?: 'episode' \| 'movie' \| 'concert' \| 'event' \| 'live-event' \| 'broadcast' \| 'tv-series' \| 'video-game' \| 'clip' \| 'vlog' \| 'audio' \| 'audio-book' \| 'music' \| 'album' \| 'articles' \| 'news' \| 'radio' \| 'podcast' \| 'book' \| 'sports-game';
205	in_languages?: string[];
206	country_of_origin?: string;
207	filter_tags?: string[];
208	hash_tags?: string[];
209	content_rating?: string[];
210	persons?: Array<{
211	name: string;
212	role: GoogleMediaPersonOrgRole;
213	custom_role?: string;
214	rank?: number;
215	uri?: string;
216	}>;
217	organizations?: Array<{
218	name: string;
219	role: GoogleMediaPersonOrgRole;
220	custom_role?: string;
221	/**
222	* Is this really a string? That's what their JSON schema says, but it's an int for `persons`...
223	*/
224	rank?: string;
225	uri?: string;
226	}>;
227	aggregate_ratings?: Array<{
228	rating_source: string;
229	rating_score?: number;
230	rating_count?: number;
231	}>
232	}
233
234	interface CreateGoogleMediaDocumentMetadataArgs {
235	post: DbPost;
236	tags?: {
237	_id: string;
238	name: string;
239	core: boolean;
240	}[];
241	authorIds?: string[];
242	}
243
244	function createGoogleMediaDocumentJson({ post, tags, authorIds }: CreateGoogleMediaDocumentMetadataArgs): protos.google.cloud.discoveryengine.v1.IDocument {
245	const tagIds = tags?.map(tag => tag._id) ?? [];	×
246	if (tagIds.length === 0) {	×
247	tagIds.push('none')	×
248	}
249	const persons = authorIds?.map(authorId => ({ name: authorId, role: 'author' } as const));	×
250
251	const metadata: GoogleMediaDocumentMetadata = {	×
252	title: post.title,
253	uri: `https://www.lesswrong.com${postGetPageUrl(post)}`,
254	available_time: post.postedAt.toISOString(),
255	categories: tagIds,
256	filter_tags: tagIds,
257	persons,
258	media_type: 'articles',
259	};
260
261	return {	×
262	id: post._id,
263	schemaId: 'default_schema',
264	jsonData: JSON.stringify(metadata),
265	};
266	}
267
268	interface InViewEvent {
269	userId: string;
270	postId: string;
271	/**
272	* When generating the table with these events, we forgot to alias the MIN(timestamp) back to `timestamp`.
273	*/
274	min: string;
275	}
276
277	function createGoogleViewItemEvent(eventType: 'view-item'\|'media-play', readStatus: DbReadStatus): protos.google.cloud.discoveryengine.v1.IUserEvent {
278	const { userId, postId, lastUpdated: timestamp } = readStatus;	×
279
280	return {	×
281	eventType,
282	// TODO - this should be clientId if doing real stuff with it
283	userPseudoId: userId,
284	eventTime: {
285	seconds: timestamp.getTime() / 1000,
286	nanos: 0
287	},
288	userInfo: { userId },
289	documents: [ { id: postId } ],
290	};
291	}
292
293	function createGoogleMediaCompleteEvent(inViewEvent: InViewEvent): protos.google.cloud.discoveryengine.v1.IUserEvent {
294	const { userId, postId, min } = inViewEvent;	×
295
296	return {	×
297	eventType: 'media-complete',
298	// TODO - this should be clientId if doing real stuff with it
299	userPseudoId: userId,
300	eventTime: {
301	seconds: new Date(min).getTime() / 1000,
302	nanos: 0
303	},
304	userInfo: { userId },
305	documents: [ { id: postId } ],
306	mediaInfo: {
307	mediaProgressPercentage: 1
308	}
309	};
310	}
311
312	function createDelimitedJsonString<T>(records: T[]) {
313	return records.map(record => JSON.stringify(record)).join('\n');	×
314	}
315
316	function indexInViewEvents(inViewEvents: InViewEvent[]): Record<string, Record<string, string>> {
317	const indexedInViewEvents: Record<string, Record<string, string>> = {};	×
318
319	// eslint-disable-next-line no-console
320	console.log(`Indexing ${inViewEvents.length} inViewEvents`);	×
321
322	inViewEvents.reduce((acc, { postId, userId, min }) => {	×
323	acc[postId] ??= {};	×
324	acc[postId][userId] = min;	×
325	return acc;	×
326	}, indexedInViewEvents);
327
328	// eslint-disable-next-line no-console
329	console.log(`Indexed inViewEvents for ${Object.keys(indexedInViewEvents).length} posts`);	×
330
331	return indexedInViewEvents;	×
332	}
333
334	function createViewHomePageEvent({userId, timestamp}: { userId: string, timestamp: string }) {
335	return {	×
336	eventType: 'view-home-page',
337	userPseudoId: userId,
338	eventTime: {
339	seconds: new Date(timestamp).getTime() / 1000,
340	nanos: 0
341	},
342	userInfo: { userId }
343	};
344	}
345
346	async function backfillPosts(offsetDate?: Date) {
347	const adminContext = createAdminContext();	×
348	const db = getSqlClientOrThrow();	×
349	const documentClient = getGoogleDocumentServiceClientOrThrow();	×
350	const userEventClient = getGoogleUserEventServiceClientOrThrow();	×
351
352	const inViewEvents: InViewEvent[] = JSON.parse((await readFile('/Users/robert/Documents/repos/ForumMagnum/in_view_events_for_google_20240415.json')).toString());	×
353	const indexedInViewEvents = indexInViewEvents(inViewEvents);	×
354
355	if (!offsetDate) {	×
356	({ offsetDate } = await db.one<{ offsetDate: Date }>('SELECT MIN("createdAt") AS "offsetDate" FROM "Posts"'));	×
357	}
358
359	// eslint-disable-next-line no-console
360	console.log(`Initial post batch offset date: ${offsetDate.toISOString()}`);	×
361
362	let batch = await getPostBatch(offsetDate);	×
363
364	try {	×
365	while (batch.length) {	×
366	// eslint-disable-next-line no-console
367	console.log(`Post batch size: ${batch.length}.`);	×
368
369	const postsWithTags = batch.map(({ tags, authors, authorIds, upvoteCount, ...post }) => ({	×
370	post,
371	tags: filterNonnull(tags.map(([_id, name, core]) => ({ _id, name, core }))),	×
372	authors,
373	authorIds,
374	upvoteCount
375	}));
376
377	const googleMediaDocuments = postsWithTags.map(createGoogleMediaDocumentJson);	×
378	if (googleMediaDocuments.length) {	×
379	const postIds = postsWithTags.map(({ post }) => post._id);	×
380	const readStatusOperation = () => ReadStatuses.find(	×
381	{ postId: { $in: postIds }, isRead: true },
382	undefined,
383	{ _id: 1, userId: 1, postId: 1, lastUpdated: 1 }
384	).fetch();
385
386	const [importDocumentsOperation] = await documentClient.importDocuments({ inlineSource: { documents: googleMediaDocuments }, parent: GOOGLE_PARENT_DOCUMENTS_PATH });	×
387	const [[importDocumentsResponse], readStatuses] = await Promise.all([	×
388	importDocumentsOperation.promise(),
389	readStatusOperation()
390	]);
391
392	if (importDocumentsResponse.errorSamples?.length) {	×
393	// eslint-disable-next-line no-console
394	console.log('Error importing documents', { error: importDocumentsResponse.errorSamples[0] });	×
395	}
396
397	const postReadStatusMap = groupBy(readStatuses, 'postId');	×
398
399	const viewItemEvents = readStatuses.map(readStatus => createGoogleViewItemEvent('view-item', readStatus));	×
400	const mediaPlayEvents = readStatuses.map(readStatus => createGoogleViewItemEvent('media-play', readStatus));	×
401
402	const inViewEvents = postsWithTags.map(({ post }) => {	×
403	const postReadStatuses = postReadStatusMap[post._id] ?? [];	×
404	const correspondingInViewEvents = postReadStatuses.filter(({ postId, userId }) => !!indexedInViewEvents[postId!]?.[userId]);	×
405	return correspondingInViewEvents.map(({ postId, userId }) => {	×
406	const min = indexedInViewEvents[postId!][userId];	×
407	return { postId, userId, min };	×
408	})
409	}).flat();
410
411	const mediaCompleteEvents = inViewEvents.map(createGoogleMediaCompleteEvent);	×
412
413	const userEvents = [...viewItemEvents, ...mediaPlayEvents, ...mediaCompleteEvents];	×
414
415	const chunkedUserEvents = chunk(userEvents, 90000);	×
416	for (const chunk of chunkedUserEvents) {	×
417	const [importUserEventsOperation] = await userEventClient.importUserEvents({ inlineSource: { userEvents: chunk }, parent: GOOGLE_PARENT_EVENTS_PATH });	×
418	const [importUserEventsResponse] = await importUserEventsOperation.promise();	×
419	if (importUserEventsResponse.errorSamples?.length) {	×
420	// eslint-disable-next-line no-console
421	console.log('Error importing user events', { error: importUserEventsResponse.errorSamples[0] });	×
422	}
423	}
424	}
425
UNCOV 426	const nextOffsetDate: Date \| undefined = getNextOffsetDate(offsetDate, batch);	×
427	if (!nextOffsetDate) {	×
428	return;	×
429	}
430	offsetDate = nextOffsetDate;	×
431	batch = await getPostBatch(offsetDate);	×
432	}
433	} catch (err) {
434	// eslint-disable-next-line no-console
435	console.log(`Error when backfilling BigQuery/Google with posts. Last offset date: ${offsetDate.toISOString()}`, { err });	×
436	}
437	}
438
439	interface FrontpageView {
440	userId: string;
441	timestamp: string;
442	}
443
444	async function backfillFrontpageViews(offsetDate?: Date) {
445	const userEventClient = getGoogleUserEventServiceClientOrThrow();	×
446
447	const frontpageViewEvensts: FrontpageView[] = JSON.parse((await readFile('/Users/rbloom/git/lesswrongSuite/logged_in_user_frontpage_loads_20240416.json')).toString());	×
448
NEW 449	const chunkedFrontpageViews = chunk(frontpageViewEvensts, 90000);	×
450
451	for (const chunk of chunkedFrontpageViews) {	×
452	// eslint-disable-next-line no-console
453	console.log(`Processing chunk of ${chunk.length} frontpage views`);	×
454	const userEvents = chunk.map(({ userId, timestamp }) => createViewHomePageEvent({ userId, timestamp }));	×
455	const [importUserEventsOperation] = await userEventClient.importUserEvents({ inlineSource: { userEvents }, parent: GOOGLE_PARENT_EVENTS_PATH });	×
456	const [importUserEventsResponse] = await importUserEventsOperation.promise();	×
457	if (importUserEventsResponse.errorSamples?.length) {	×
458	// eslint-disable-next-line no-console
459	console.log('Error importing frontpage loads', { error: importUserEventsResponse.errorSamples[0] });	×
460	}
461	}
462	}
463
464
465	Globals.backfillBigQueryPosts = backfillPosts;	18✔
466	Globals.backfillFrontpageViews = backfillFrontpageViews;	18✔

ForumMagnum / ForumMagnum / 8730738677

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