8711651453

Committed 16 Apr 2024 07:51PM UTC coverage: 31.922% (-0.03%) from 31.951%

Build # 8711651453

Build Type

push

github

Committed by

darkruby501

Commit Message

backfill script mostly working

Run Details

4628 of 37858 branches covered (12.22%)

Branch coverage included in aggregate %.

3 of 21 new or added lines in 1 file covered. (14.29%)

3 existing lines in 2 files now uncovered.

27047 of 61368 relevant lines covered (44.07%)

182.45 hits per line

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

10.8

/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/lw-recommendations-2/locations/global/dataStores/datastore-lwrecommendations2_1713296505582/branches/default_branch';
const GOOGLE_PARENT_EVENTS_PATH = 'projects/lw-recommendations-2/locations/global/dataStores/datastore-lwrecommendations2_1713296505582';

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;

  if (!(!!userId && !!postId && !!timestamp)) {
    console.log('something missing in createGoogleViewItemEvent', { eventType, userId, postId, timestamp });
  }

  if (postId!.length >= 128) {
    console.log('postId too long', { postId });
  }

  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, timestamp } = inViewEvent;

  if (!(!!userId && !!postId && !!timestamp)) {
    console.log('something missing in createGoogleMediaCompleteEvent', { userId, postId, timestamp });
  }
  
  if (postId!.length >= 128) {
    console.log('postId too long', { postId });
  }

  return {
    eventType: 'media-complete',
    // 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 } ],
    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/rbloom/git/lesswrongSuite/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_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] ?? [];
          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];

        //chunk sending user events with batch size of 90000, using lodash chunk function
        const chunkSize = 90000;
        const chunkedUserEvents = chunk(userEvents, chunkSize);
        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 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 BigQuery/Google 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	import chunk from "lodash/chunk";	18✔
12
13	const GOOGLE_PARENT_DOCUMENTS_PATH = 'projects/lw-recommendations-2/locations/global/dataStores/datastore-lwrecommendations2_1713296505582/branches/default_branch';	18✔
14	const GOOGLE_PARENT_EVENTS_PATH = 'projects/lw-recommendations-2/locations/global/dataStores/datastore-lwrecommendations2_1713296505582';	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	const persons = authorIds?.map(authorId => ({ name: authorId, role: 'author' } as const));	×
247
248	const metadata: GoogleMediaDocumentMetadata = {	×
249	title: post.title,
250	uri: `https://www.lesswrong.com${postGetPageUrl(post)}`,
251	available_time: post.postedAt.toISOString(),
252	categories: tagIds ?? [],	×
253	filter_tags: tagIds,
254	persons,
255	media_type: 'articles',
256	};
257
258	return {	×
259	id: post._id,
260	schemaId: 'default_schema',
261	jsonData: JSON.stringify(metadata),
262	// content: {
263	// mimeType: 'text/html',
264	// rawBytes: Buffer.from(post.contents?.html ?? '').toString('base64')
265	// }
266	};
267	}
268
269	interface InViewEvent {
270	userId: string;
271	postId: string;
272	timestamp: Date;
273	}
274
275	function createGoogleViewItemEvent(eventType: 'view-item'\|'media-play', readStatus: DbReadStatus): protos.google.cloud.discoveryengine.v1.IUserEvent {
276	const { userId, postId, lastUpdated: timestamp } = readStatus;	×
277
NEW 278	if (!(!!userId && !!postId && !!timestamp)) {	×
NEW 279	console.log('something missing in createGoogleViewItemEvent', { eventType, userId, postId, timestamp });	×
280	}
281
NEW 282	if (postId!.length >= 128) {	×
NEW 283	console.log('postId too long', { postId });	×
284	}
285
UNCOV 286	return {	×
287	eventType,
288	// TODO - this should be clientId if doing real stuff with it
289	userPseudoId: userId,
290	eventTime: {
291	seconds: timestamp.getTime() / 1000,
292	nanos: 0
293	},
294	userInfo: { userId },
295	documents: [ { id: postId } ],
296	};
297	}
298
299	function createGoogleMediaCompleteEvent(inViewEvent: InViewEvent): protos.google.cloud.discoveryengine.v1.IUserEvent {
300	const { userId, postId, timestamp } = inViewEvent;	×
301
NEW 302	if (!(!!userId && !!postId && !!timestamp)) {	×
NEW 303	console.log('something missing in createGoogleMediaCompleteEvent', { userId, postId, timestamp });	×
304	}
305
NEW 306	if (postId!.length >= 128) {	×
NEW 307	console.log('postId too long', { postId });	×
308	}
309
UNCOV 310	return {	×
311	eventType: 'media-complete',
312	// TODO - this should be clientId if doing real stuff with it
313	userPseudoId: userId,
314	eventTime: {
315	seconds: timestamp.getTime() / 1000,
316	nanos: 0
317	},
318	userInfo: { userId },
319	documents: [ { id: postId } ],
320	mediaInfo: {
321	mediaProgressPercentage: 1
322	}
323	};
324	}
325
326	function createDelimitedJsonString<T>(records: T[]) {
327	return records.map(record => JSON.stringify(record)).join('\n');	×
328	}
329
330	// function getPostUserTimestamps(inViewTimestamps: InViewTimestamps[]): Record<string,Record<string, Date[]>>{
331	// const indexedInViewEvents: Record<string, Record<string, Date[]>> = {}
332
333	// return inViewTimestamps.reduce((acc, { postId, userId, timestamps }) => {
334	// acc[postId] ??= {};
335	// acc[postId][userId] = timestamps;
336	// return acc;
337	// }, indexedInViewEvents);
338	// }
339
340	function indexInViewEvents(inViewEvents: InViewEvent[]): Record<string, Record<string, Date>> {
341	const indexedInViewEvents: Record<string, Record<string, Date>> = {};	×
342
343	// eslint-disable-next-line no-console
344	console.log(`Indexing ${inViewEvents.length} inViewEvents`);	×
345
346	inViewEvents.reduce((acc, { postId, userId, timestamp }) => {	×
347	acc[postId] ??= {};	×
348	acc[postId][userId] = timestamp;	×
349	return acc;	×
350	}, indexedInViewEvents);
351
352	// eslint-disable-next-line no-console
353	console.log(`Indexed inViewEvents for ${Object.keys(indexedInViewEvents).length} posts`);	×
354
355	return indexedInViewEvents;	×
356	}
357
358
359	async function backfillPosts(offsetDate?: Date) {
360	const adminContext = createAdminContext();	×
361	const db = getSqlClientOrThrow();	×
362	const documentClient = getGoogleDocumentServiceClientOrThrow();	×
363	const userEventClient = getGoogleUserEventServiceClientOrThrow();	×
364
NEW 365	const inViewEvents: InViewEvent[] = JSON.parse((await readFile('/Users/rbloom/git/lesswrongSuite/in_view_events_for_google_20240415.json')).toString());	×
366	const indexedInViewEvents = indexInViewEvents(inViewEvents);	×
367
368	if (!offsetDate) {	×
369	({ offsetDate } = await db.one<{ offsetDate: Date }>('SELECT MIN("createdAt") AS "offsetDate" FROM "Posts"'));	×
370	}
371
372	// eslint-disable-next-line no-console
373	console.log(`Initial post batch offset date: ${offsetDate.toISOString()}`);	×
374
375	let batch = await getPostBatch(offsetDate);	×
376
377	try {	×
378	while (batch.length) {	×
379	// eslint-disable-next-line no-console
380	console.log(`Post batch size: ${batch.length}.`);	×
381
382	const postsWithTags = batch.map(({ tags, authors, authorIds, upvoteCount, ...post }) => ({	×
383	post,
384	tags: filterNonnull(tags.map(([_id, name, core]) => ({ _id, name, core }))),	×
385	authors,
386	authorIds,
387	upvoteCount
388	}));
389
390	const googleMediaDocuments = postsWithTags.filter(({ tags }) => tags.length).map(createGoogleMediaDocumentJson);	×
391	if (googleMediaDocuments.length) {	×
392	const postIds = postsWithTags.map(({ post }) => post._id);	×
393	const readStatusOperation = () => ReadStatuses.find(	×
394	{ postId: { $in: postIds }, isRead: true },
395	undefined,
396	{ _id: 1, userId: 1, postId: 1, lastUpdated: 1 }
397	).fetch();
398
NEW 399	const [importDocumentsOperation] = await documentClient.importDocuments({ inlineSource: { documents: googleMediaDocuments }, parent: GOOGLE_PARENT_DOCUMENTS_PATH });	×
400	const [[importDocumentsResponse], readStatuses] = await Promise.all([	×
401	importDocumentsOperation.promise(),
402	readStatusOperation()
403	]);
404
405	if (importDocumentsResponse.errorSamples?.length) {	×
406	// eslint-disable-next-line no-console
407	console.log('Error importing documents', { error: importDocumentsResponse.errorSamples[0] });	×
408	}
409
410	const postReadStatusMap = groupBy(readStatuses, 'postId');	×
411
412	const viewItemEvents = readStatuses.map(readStatus => createGoogleViewItemEvent('view-item', readStatus));	×
413	const mediaPlayEvents = readStatuses.map(readStatus => createGoogleViewItemEvent('media-play', readStatus));	×
414
415	const inViewEvents = postsWithTags.map(({ post }) => {	×
416	const postReadStatuses = postReadStatusMap[post._id] ?? [];	×
417	return postReadStatuses.filter(({ postId, userId }) => !!indexedInViewEvents[postId!]?.[userId]).map(({ postId, userId }) => {	×
418	const timestamp = indexedInViewEvents[postId!][userId];	×
419	return { postId, userId, timestamp };	×
420	})
421	}).flat();
422
423	const mediaCompleteEvents = inViewEvents.map(createGoogleMediaCompleteEvent);	×
424
425	const userEvents = [...viewItemEvents, ...mediaPlayEvents, ...mediaCompleteEvents];	×
426
427	//chunk sending user events with batch size of 90000, using lodash chunk function
NEW 428	const chunkSize = 90000;	×
NEW 429	const chunkedUserEvents = chunk(userEvents, chunkSize);	×
NEW 430	for (const chunk of chunkedUserEvents) {	×
NEW 431	const [importUserEventsOperation] = await userEventClient.importUserEvents({ inlineSource: { userEvents: chunk }, parent: GOOGLE_PARENT_EVENTS_PATH });	×
NEW 432	const [importUserEventsResponse] = await importUserEventsOperation.promise();	×
NEW 433	if (importUserEventsResponse.errorSamples?.length) {	×
434	// eslint-disable-next-line no-console
NEW 435	console.log('Error importing user events', { error: importUserEventsResponse.errorSamples[0] });	×
436	}
437	}
438	}
439	// const bigQueryRecordBatch = await Promise.all(postsWithTags.map(({ post, tags, authors, upvoteCount }) => createAEPostRecord({ post, context: adminContext, tags, authors, upvoteCount })));
440	// await writeFile(`aestudios/lw_posts_${offsetDate.toISOString()}.json`, JSON.stringify(bigQueryRecordBatch));
441
442	// const bigQueryRecordBatch = await executePromiseQueue(postsWithTags.map(({ post, tags }) => () => createBigQueryPostRecord({ post, context: adminContext, tags })), 10);
443	// await writeFile(`delimited_bigquery_posts_${offsetDate.toISOString()}.json`, createDelimitedJsonString(bigQueryRecordBatch));
444
445	const nextOffsetDate: Date \| undefined = getNextOffsetDate(offsetDate, batch);	×
446	if (!nextOffsetDate) {	×
447	return;	×
448	}
449	offsetDate = nextOffsetDate;	×
450	batch = await getPostBatch(offsetDate);	×
451	}
452	} catch (err) {
453	// eslint-disable-next-line no-console
NEW 454	console.log(`Error when backfilling BigQuery/Google with posts. Last offset date: ${offsetDate.toISOString()}`, { err });	×
455	}
456	}
457
458	Globals.backfillBigQueryPosts = backfillPosts;	18✔

ForumMagnum / ForumMagnum / 8711651453

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