8698057045

Committed 16 Apr 2024 12:35AM UTC coverage: 31.937% (-0.007%) from 31.944%

Build # 8698057045

Build Type

push

github

Committed by

darkruby501

Commit Message

progress on uploading to google

Run Details

4630 of 37842 branches covered (12.24%)

Branch coverage included in aggregate %.

7 of 34 new or added lines in 1 file covered. (20.59%)

2 existing lines in 1 file now uncovered.

27048 of 61348 relevant lines covered (44.09%)

182.5 hits per line

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

13.57

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

import { 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');
}

const testInViewEvents = [
  {
    userId: 'abc123',
    postId: 'xyz789',
    timestamp: new Date(),
  },
  {
    userId: 'def456',
    postId: 'xyz789',
    timestamp: new Date(),
  }
]

interface InViewTimestamps {
  postId: string;
  userId: string;
  timestamps: Date[];
}

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);
}


const indexedInViewEvents = groupBy(testInViewEvents, 'postId');

function getInViewEventsByPostId(postId: string) {  
  return indexedInViewEvents[postId] ?? [];
}

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

  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 [operation] = await documentClient.importDocuments({ inlineSource: { documents: googleMediaDocuments }, parent: GOOGLE_PARENT_PATH });
        const [response] = await operation.promise();
        if (response.errorSamples?.length) {
          // eslint-disable-next-line no-console
          console.log({ error: response.errorSamples[0] });
        }  

        const postIds = postsWithTags.map(({post}) => post._id)

        const readStatuses = await ReadStatuses.find(
          { postId: { $in: postIds }, isRead: true }, 
          undefined, 
          { _id: 1, userId: 1, postId: 1, lastUpdated: 1 }
        ).fetch()
        

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

        const inViewEvents = postsWithTags.map(({ post }) => getInViewEventsByPostId(post._id)).flat();
        const mediaCompleteEvents = inViewEvents.map(createGoogleMediaCompleteEvent);

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

        const [userEventsOperation] = await userEventClient.importUserEvents({ inlineSource: { userEvents }, parent: GOOGLE_PARENT_PATH });
        const [userEventsResponse] = await userEventsOperation.promise();
        if (userEventsResponse.errorSamples?.length) {
          // eslint-disable-next-line no-console
          console.log({ error: userEventsResponse.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 { 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) {	×
UNCOV 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✔
NEW 31	if (!client) {	×
NEW 32	client = new UserEventServiceClient();	×
33	}
34
NEW 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 {
NEW 274	const { userId, postId, lastUpdated: timestamp } = readStatus;	×
275
NEW 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 {
NEW 289	const { userId, postId, timestamp } = inViewEvent;	×
290
NEW 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	const testInViewEvents = [	18✔
311	{
312	userId: 'abc123',
313	postId: 'xyz789',
314	timestamp: new Date(),
315	},
316	{
317	userId: 'def456',
318	postId: 'xyz789',
319	timestamp: new Date(),
320	}
321	]
322
323	interface InViewTimestamps {
324	postId: string;
325	userId: string;
326	timestamps: Date[];
327	}
328
329	function getPostUserTimestamps(inViewTimestamps: InViewTimestamps[]): Record<string,Record<string, Date[]>>{
NEW 330	const indexedInViewEvents: Record<string, Record<string, Date[]>> = {}	×
331
NEW 332	return inViewTimestamps.reduce((acc, { postId, userId, timestamps }) => {	×
NEW 333	acc[postId] ??= {};	×
NEW 334	acc[postId][userId] = timestamps;	×
NEW 335	return acc;	×
336	}, indexedInViewEvents);
337	}
338
339
340	const indexedInViewEvents = groupBy(testInViewEvents, 'postId');	18✔
341
342	function getInViewEventsByPostId(postId: string) {
NEW 343	return indexedInViewEvents[postId] ?? [];	×
344	}
345
346	async function backfillPosts(offsetDate?: Date) {
347	const adminContext = createAdminContext();	×
348	const db = getSqlClientOrThrow();	×
NEW 349	const documentClient = getGoogleDocumentServiceClientOrThrow();	×
NEW 350	const userEventClient = getGoogleUserEventServiceClientOrThrow();	×
351
352	if (!offsetDate) {	×
353	({ offsetDate } = await db.one<{ offsetDate: Date }>('SELECT MIN("createdAt") AS "offsetDate" FROM "Posts"'));	×
354	}
355
356	// eslint-disable-next-line no-console
357	console.log(`Initial post batch offset date: ${offsetDate.toISOString()}`);	×
358
359	let batch = await getPostBatch(offsetDate);	×
360
361	try {	×
362	while (batch.length) {	×
363	// eslint-disable-next-line no-console
364	console.log(`Post batch size: ${batch.length}.`);	×
365
366	const postsWithTags = batch.map(({ tags, authors, authorIds, upvoteCount, ...post }) => ({	×
367	post,
368	tags: filterNonnull(tags.map(([_id, name, core]) => ({ _id, name, core }))),	×
369	authors,
370	authorIds,
371	upvoteCount
372	}));
373
374	const googleMediaDocuments = postsWithTags.filter(({ tags }) => tags.length).map(createGoogleMediaDocumentJson);	×
375	if (googleMediaDocuments.length) {	×
NEW 376	const [operation] = await documentClient.importDocuments({ inlineSource: { documents: googleMediaDocuments }, parent: GOOGLE_PARENT_PATH });	×
377	const [response] = await operation.promise();	×
378	if (response.errorSamples?.length) {	×
379	// eslint-disable-next-line no-console
UNCOV 380	console.log({ error: response.errorSamples[0] });	×
381	}
382
NEW 383	const postIds = postsWithTags.map(({post}) => post._id)	×
384
NEW 385	const readStatuses = await ReadStatuses.find(	×
386	{ postId: { $in: postIds }, isRead: true },
387	undefined,
388	{ _id: 1, userId: 1, postId: 1, lastUpdated: 1 }
389	).fetch()
390
391
NEW 392	const viewItemEvents = readStatuses.map(readStatus => createGoogleViewItemEvent('view-item', readStatus));	×
NEW 393	const mediaPlayEvents = readStatuses.map(readStatus => createGoogleViewItemEvent('media-play', readStatus));	×
394
NEW 395	const inViewEvents = postsWithTags.map(({ post }) => getInViewEventsByPostId(post._id)).flat();	×
NEW 396	const mediaCompleteEvents = inViewEvents.map(createGoogleMediaCompleteEvent);	×
397
NEW 398	const userEvents = [...viewItemEvents, ...mediaPlayEvents, ...mediaCompleteEvents];	×
399
NEW 400	const [userEventsOperation] = await userEventClient.importUserEvents({ inlineSource: { userEvents }, parent: GOOGLE_PARENT_PATH });	×
NEW 401	const [userEventsResponse] = await userEventsOperation.promise();	×
NEW 402	if (userEventsResponse.errorSamples?.length) {	×
403	// eslint-disable-next-line no-console
NEW 404	console.log({ error: userEventsResponse.errorSamples[0] });	×
405	}
406
407
408
409
410	}
411	// const bigQueryRecordBatch = await Promise.all(postsWithTags.map(({ post, tags, authors, upvoteCount }) => createAEPostRecord({ post, context: adminContext, tags, authors, upvoteCount })));
412	// await writeFile(`aestudios/lw_posts_${offsetDate.toISOString()}.json`, JSON.stringify(bigQueryRecordBatch));
413
414	// const bigQueryRecordBatch = await executePromiseQueue(postsWithTags.map(({ post, tags }) => () => createBigQueryPostRecord({ post, context: adminContext, tags })), 10);
415	// await writeFile(`delimited_bigquery_posts_${offsetDate.toISOString()}.json`, createDelimitedJsonString(bigQueryRecordBatch));
416
417	const nextOffsetDate: Date \| undefined = getNextOffsetDate(offsetDate, batch);	×
418	if (!nextOffsetDate) {	×
419	return;	×
420	}
421	offsetDate = nextOffsetDate;	×
422	batch = await getPostBatch(offsetDate);	×
423	}
424	} catch (err) {
425	// eslint-disable-next-line no-console
426	console.log(`Error when backfilling recombee with posts. Last offset date: ${offsetDate.toISOString()}`, { err });	×
427	}
428	}
429
430	Globals.backfillBigQueryPosts = backfillPosts;	18✔

ForumMagnum / ForumMagnum / 8698057045

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