8698982111

Committed 16 Apr 2024 02:18AM UTC coverage: 31.951% (+0.01%) from 31.937%

Build # 8698982111

Build Type

push

github

Committed by

b0b3rt

Commit Message

user event import wip

Run Details

4630 of 37832 branches covered (12.24%)

Branch coverage included in aggregate %.

1 of 25 new or added lines in 1 file covered. (4.0%)

1 existing line in 1 file now uncovered.

27046 of 61307 relevant lines covered (44.12%)

182.63 hits per line

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

11.41

/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";

const GOOGLE_PARENT_PATH = 'projects/lesswrong-recommendations/locations/global/dataStores/datastore1_1712780871028/branches/default_branch';

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);
  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),
    // content: {
    //   mimeType: 'text/html',
    //   rawBytes: Buffer.from(post.contents?.html ?? '').toString('base64')
    // }
  };
}

interface InViewEvent {
  userId: string;
  postId: string;
  timestamp: Date;
}

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
    },
    userInfo: { userId },
    documents: [ { id: postId } ],
  };
}

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

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

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

// function getPostUserTimestamps(inViewTimestamps: InViewTimestamps[]): Record<string,Record<string, Date[]>>{
//   const indexedInViewEvents: Record<string, Record<string, Date[]>> = {}

//   return inViewTimestamps.reduce((acc, { postId, userId, timestamps }) => {
//     acc[postId] ??= {};
//     acc[postId][userId] = timestamps;
//     return acc;
//   }, indexedInViewEvents);
// }

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

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

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

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

  return indexedInViewEvents;
}


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.filter(({ tags }) => tags.length).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_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] ?? [];
          return postReadStatuses.filter(({ postId, userId }) => !!indexedInViewEvents[postId!]?.[userId]).map(({ postId, userId }) => {
            const timestamp = indexedInViewEvents[postId!][userId];
            return { postId, userId, timestamp };
          })
        }).flat();

        const mediaCompleteEvents = inViewEvents.map(createGoogleMediaCompleteEvent);

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

        const [importUserEventsOperation] = await userEventClient.importUserEvents({ inlineSource: { userEvents }, parent: GOOGLE_PARENT_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 bigQueryRecordBatch = await Promise.all(postsWithTags.map(({ post, tags, authors, upvoteCount }) => createAEPostRecord({ post, context: adminContext, tags, authors, upvoteCount })));
      // await writeFile(`aestudios/lw_posts_${offsetDate.toISOString()}.json`, JSON.stringify(bigQueryRecordBatch));
      
      // const bigQueryRecordBatch = await executePromiseQueue(postsWithTags.map(({ post, tags }) => () => createBigQueryPostRecord({ post, context: adminContext, tags })), 10);
      // await writeFile(`delimited_bigquery_posts_${offsetDate.toISOString()}.json`, createDelimitedJsonString(bigQueryRecordBatch));
  
      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 recombee with posts.  Last offset date: ${offsetDate.toISOString()}`, { err });
  }
}

Globals.backfillBigQueryPosts = backfillPosts;

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

ForumMagnum / ForumMagnum / 8698982111

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