From 3b0fe18b9b5cda9c75a4f732639180814ef8bdf0 Mon Sep 17 00:00:00 2001 From: Battlesquid <25509915+Battlesquid@users.noreply.github.com> Date: Thu, 28 May 2026 09:37:47 -0700 Subject: [PATCH 01/10] fix: add crosspost flag check --- packages/store/src/database.ts | 2 +- services/bot/src/broadcaster.ts | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/store/src/database.ts b/packages/store/src/database.ts index ab1f4bb..f3e2b50 100644 --- a/packages/store/src/database.ts +++ b/packages/store/src/database.ts @@ -189,7 +189,7 @@ export const insertReplayEvents = async ( event: EventQueueType.Replay, payload: { question }, })); - return trycatch(() => d.insert(schema.event_queue).values(events).onConflictDoNothing()); + return trycatch(() => d.insert(schema.event_queue).values(events).onConflictDoNothing().returning()); }; export const clearReplayEvents = async (d: PostgresJsDatabase = db()) => { diff --git a/services/bot/src/broadcaster.ts b/services/bot/src/broadcaster.ts index f2bd07b..76f6a62 100644 --- a/services/bot/src/broadcaster.ts +++ b/services/bot/src/broadcaster.ts @@ -8,7 +8,7 @@ import { import { chunk, entries, groupby, trycatch } from "@qnaplus/utils"; import { container } from "@sapphire/framework"; import { Cron } from "croner"; -import { ChannelType, type EmbedBuilder, type NewsChannel, channelMention } from "discord.js"; +import { ChannelType, type EmbedBuilder, MessageFlags, type NewsChannel, channelMention } from "discord.js"; import type { Logger } from "pino"; import { buildEventEmbed } from "./formatting"; import type { PinoLoggerAdapter } from "./utils/logger_adapter"; @@ -39,7 +39,7 @@ const getChannel = (program: string) => { const broadcast = async (channel: NewsChannel, embeds: EmbedBuilder[]) => { const message = await channel.send({ embeds }); - if (message.crosspostable) { + if (message.crosspostable && !message.flags.has(MessageFlags.Crossposted)) { await message.crosspost(); } }; From 9166b43a4ee4a9658c703554251b82064ae4294f Mon Sep 17 00:00:00 2001 From: Battlesquid Date: Thu, 4 Jun 2026 22:47:37 -0700 Subject: [PATCH 02/10] fix: only update storage if updates detected --- services/updater/src/index.ts | 46 +++++++++++---------- services/updater/src/update_database.ts | 6 +-- services/updater/src/update_forum_status.ts | 6 +-- 3 files changed, 30 insertions(+), 28 deletions(-) diff --git a/services/updater/src/index.ts b/services/updater/src/index.ts index 1f423c5..043a57b 100644 --- a/services/updater/src/index.ts +++ b/services/updater/src/index.ts @@ -10,36 +10,38 @@ import { updateForumStatus } from "./update_forum_status"; import { updateStorage } from "./update_storage"; const update = async (client: FetchClient, logger: Logger) => { - await updateDatabase(client, logger); - await updateStorage(logger); - await updateForumStatus(client, logger); + const { hasUpdates } = await updateDatabase(client, logger); + if (hasUpdates) { + await updateStorage(logger); + } + await updateForumStatus(client, logger); }; const start = async (client: FetchClient, logger: Logger) => { - logger.info("Starting database update job"); - const job = new Cron(getenv("DATABASE_UPDATE_INTERVAL"), () => update(client, logger), { - name: "updater", - protect: true, - catch(error) { - logger.error({ error }, "An error occurred while updating database."); - }, - }); - job.trigger(); + logger.info("Starting database update job"); + const job = new Cron(getenv("DATABASE_UPDATE_INTERVAL"), () => update(client, logger), { + name: "updater", + protect: true, + catch(error) { + logger.error({ error }, "An error occurred while updating database."); + }, + }); + job.trigger(); }; (async () => { - await initializeEnv(); + await initializeEnv(); - const logger = getLoggerInstance("qnaplus-updater"); - logger.info("Starting updater service"); + const logger = getLoggerInstance("qnaplus-updater"); + logger.info("Starting updater service"); - const [error] = await testConnection(); - if (error) { - logger.error({ error }, "Unable to establish database connection, exiting."); - process.exit(1); - } + const [error] = await testConnection(); + if (error) { + logger.error({ error }, "Unable to establish database connection, exiting."); + process.exit(1); + } - const client = new CurlImpersonateScrapingClient(logger); + const client = new CurlImpersonateScrapingClient(logger); - start(client, logger); + start(client, logger); })(); diff --git a/services/updater/src/update_database.ts b/services/updater/src/update_database.ts index f57d58c..e58c097 100644 --- a/services/updater/src/update_database.ts +++ b/services/updater/src/update_database.ts @@ -7,7 +7,7 @@ import { getMetadata, updateQuestions } from "@qnaplus/store"; import type { Logger } from "pino"; export interface DatabaseUpdateStatus { - updateStorage: boolean; + hasUpdates: boolean; } export const updateDatabase = async ( @@ -16,7 +16,7 @@ export const updateDatabase = async ( ): Promise => { const logger = _logger?.child({ label: "update_database" }); const status: DatabaseUpdateStatus = { - updateStorage: false, + hasUpdates: false, }; logger.info("Starting database update."); const [metadataError, metadata] = await getMetadata(); @@ -50,7 +50,7 @@ export const updateDatabase = async ( ); return status; } - status.updateStorage = updates.length !== 0; + status.hasUpdates = updates.length !== 0; logger.info(`${updates.length} new updates detected.`); return status; diff --git a/services/updater/src/update_forum_status.ts b/services/updater/src/update_forum_status.ts index 357a005..44abee4 100644 --- a/services/updater/src/update_forum_status.ts +++ b/services/updater/src/update_forum_status.ts @@ -86,16 +86,16 @@ export const updateForumStatus = async ( newStates.push({ program, open }); } if (newStates.length === 0) { - logger.warn("Unable to update program statings."); + logger.warn("Unable to update forum states."); return; } const [updatedError] = await updateForumStates(newStates); if (updatedError) { logger.error( { error: updatedError }, - "An error occurred while updating program states, exiting.", + "An error occurred while updating forum states, exiting.", ); return; } - logger.info("Completed programs update."); + logger.info("Completed forum states update."); }; From 7a88b0e9486dff434393d67eaf5b0c776775f818 Mon Sep 17 00:00:00 2001 From: Battlesquid Date: Thu, 4 Jun 2026 23:21:41 -0700 Subject: [PATCH 03/10] refactor: only fetch metadata once --- packages/store/src/database.ts | 296 ++++++++++---------- packages/store/src/schema_types.ts | 4 +- packages/utils/src/variables.ts | 6 + services/updater/src/index.ts | 14 +- services/updater/src/update_database.ts | 75 ++--- services/updater/src/update_forum_status.ts | 162 +++++------ services/updater/src/update_storage.ts | 83 +++--- 7 files changed, 315 insertions(+), 325 deletions(-) diff --git a/packages/store/src/database.ts b/packages/store/src/database.ts index f3e2b50..5ff8b40 100644 --- a/packages/store/src/database.ts +++ b/packages/store/src/database.ts @@ -11,217 +11,221 @@ const pg = lazy(() => postgres(getenv("SUPABASE_TRANSACTION_URL"))); export const db = lazy(() => drizzle({ schema, client: pg() })); export const disconnectPgClient = async () => { - const client = pg(); - if (client !== null) { - await client.end(); - } + const client = pg(); + if (client !== null) { + await client.end(); + } }; export const METADATA_ROW_ID = 0; export const testConnection = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => d.execute(sql`select 1`)); + return trycatch(() => d.execute(sql`select 1`)); }; export const getQuestion = async ( - id: Question["id"], - d: PostgresJsDatabase = db(), + id: Question["id"], + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => d.query.questions.findFirst({ where: eq(schema.questions.id, id) })); + return trycatch(() => d.query.questions.findFirst({ where: eq(schema.questions.id, id) })); }; export const getAllQuestions = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => d.select().from(schema.questions)); + return trycatch(() => d.select().from(schema.questions)); }; export const getAllSeasonQuestions = async (d: PostgresJsDatabase = db()) => { - const currentSeason = d.$with("current_season").as( - d - .select({ season: sql`${schema.metadata.currentSeason}`.as("s") }) - .from(schema.metadata) - .limit(1), - ); - return trycatch(() => - d - .with(currentSeason) - .select(getTableColumns(schema.questions)) - .from(schema.questions) - .innerJoin(currentSeason, eq(schema.questions.season, currentSeason.season)), - ); + const currentSeason = d.$with("current_season").as( + d + .select({ season: sql`${schema.metadata.currentSeason}`.as("s") }) + .from(schema.metadata) + .limit(1), + ); + return trycatch(() => + d + .with(currentSeason) + .select(getTableColumns(schema.questions)) + .from(schema.questions) + .innerJoin(currentSeason, eq(schema.questions.season, currentSeason.season)), + ); }; export const getAnsweredQuestionsNewerThanDate = async ( - ms: number, - d: PostgresJsDatabase = db(), + ms: number, + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => - d - .select() - .from(schema.questions) - .where( - and( - gte(schema.questions.answeredTimestampMs, ms), - eq(schema.questions.answered, true), - ), - ), - ); + return trycatch(() => + d + .select() + .from(schema.questions) + .where( + and( + gte(schema.questions.answeredTimestampMs, ms), + eq(schema.questions.answered, true), + ), + ), + ); }; export const insertQuestions = async ( - data: Question[], - d: PostgresJsDatabase = db(), + data: Question[], + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => d.insert(schema.questions).values(data)); + return trycatch(() => d.insert(schema.questions).values(data)); }; const EXCLUDED_QUESTION = { - id: sql`excluded.id`, - url: sql`excluded.url`, - author: sql`excluded.author`, - program: sql`excluded.program`, - title: sql`excluded.title`, - question: sql`excluded.question`, - questionRaw: sql`excluded."questionRaw"`, - answer: sql`excluded.answer`, - answerRaw: sql`excluded."answerRaw"`, - season: sql`excluded.season`, - askedTimestamp: sql`excluded."askedTimestamp"`, - askedTimestampMs: sql`excluded."askedTimestampMs"`, - answeredTimestamp: sql`excluded."answeredTimestamp"`, - answeredTimestampMs: sql`excluded."answeredTimestampMs"`, - answered: sql`excluded.answered`, - tags: sql`excluded.tags`, + id: sql`excluded.id`, + url: sql`excluded.url`, + author: sql`excluded.author`, + program: sql`excluded.program`, + title: sql`excluded.title`, + question: sql`excluded.question`, + questionRaw: sql`excluded."questionRaw"`, + answer: sql`excluded.answer`, + answerRaw: sql`excluded."answerRaw"`, + season: sql`excluded.season`, + askedTimestamp: sql`excluded."askedTimestamp"`, + askedTimestampMs: sql`excluded."askedTimestampMs"`, + answeredTimestamp: sql`excluded."answeredTimestamp"`, + answeredTimestampMs: sql`excluded."answeredTimestampMs"`, + answered: sql`excluded.answered`, + tags: sql`excluded.tags`, }; const QUESTION_UPDATED_QUERY = or( - sql`${schema.questions.question} != excluded.question`, - sql`${schema.questions.answer} IS DISTINCT FROM excluded.answer`, - sql`${schema.questions.answered} != excluded.answered`, + sql`${schema.questions.question} != excluded.question`, + sql`${schema.questions.answer} IS DISTINCT FROM excluded.answer`, + sql`${schema.questions.answered} != excluded.answered`, ); export const updateQuestions = async ( - data: Question[], - d: PostgresJsDatabase = db(), + data: Question[], + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => - d - .insert(schema.questions) - .values(data) - .onConflictDoUpdate({ - target: schema.questions.id, - set: EXCLUDED_QUESTION, - setWhere: QUESTION_UPDATED_QUERY, - }) - .returning(), - ); + return trycatch(() => + d + .insert(schema.questions) + .values(data) + .onConflictDoUpdate({ + target: schema.questions.id, + set: EXCLUDED_QUESTION, + setWhere: QUESTION_UPDATED_QUERY, + }) + .returning(), + ); }; export const getEventQueue = async (d: PostgresJsDatabase = db()) => { - const formattedPayloads = d.$with("formatted_payloads").as( - d - .select({ - event: schema.event_queue.event, - payload: - sql`jsonb_build_object('id', ${schema.event_queue.id}, 'payload', ${schema.event_queue.payload})`.as( - "payload", - ), - }) - .from(schema.event_queue) - .orderBy(schema.event_queue.timestamp), - ); - - const aggregatedPayloads = d.$with("aggregated_payloads").as( - d - .select({ - event: formattedPayloads.event, - payloads: sql`array_agg(${formattedPayloads.payload})`.as("payloads"), - }) - .from(formattedPayloads) - .groupBy(formattedPayloads.event), - ); - return trycatch(() => - d - .with(formattedPayloads, aggregatedPayloads) - .select({ - queue: sql`jsonb_object_agg(${schema.event_queue.event}, ${aggregatedPayloads.payloads})`, - }) - .from(aggregatedPayloads), - ); + const formattedPayloads = d.$with("formatted_payloads").as( + d + .select({ + event: schema.event_queue.event, + payload: + sql`jsonb_build_object('id', ${schema.event_queue.id}, 'payload', ${schema.event_queue.payload})`.as( + "payload", + ), + }) + .from(schema.event_queue) + .orderBy(schema.event_queue.timestamp), + ); + + const aggregatedPayloads = d.$with("aggregated_payloads").as( + d + .select({ + event: formattedPayloads.event, + payloads: sql`array_agg(${formattedPayloads.payload})`.as("payloads"), + }) + .from(formattedPayloads) + .groupBy(formattedPayloads.event), + ); + return trycatch(() => + d + .with(formattedPayloads, aggregatedPayloads) + .select({ + queue: sql`jsonb_object_agg(${schema.event_queue.event}, ${aggregatedPayloads.payloads})`, + }) + .from(aggregatedPayloads), + ); }; export const clearEventQueue = async ( - ids: string[], - d: PostgresJsDatabase = db(), + ids: string[], + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => d.delete(schema.event_queue).where(inArray(schema.event_queue.id, ids))); + return trycatch(() => d.delete(schema.event_queue).where(inArray(schema.event_queue.id, ids))); }; export const getMetadata = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => d.query.metadata.findFirst()); + return trycatch(() => d.query.metadata.findFirst()); +}; + +export const getMetadataCurrentSeason = async (d: PostgresJsDatabase = db()) => { + return trycatch(() => d.query.metadata.findFirst({ columns: { currentSeason: true } })); }; export const updateMetadata = async ( - data: Partial>, - d: PostgresJsDatabase = db(), + data: Partial>, + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => - d - .update(schema.metadata) - .set({ ...data }) - .where(eq(schema.metadata.id, METADATA_ROW_ID)), - ); + return trycatch(() => + d + .update(schema.metadata) + .set({ ...data }) + .where(eq(schema.metadata.id, METADATA_ROW_ID)), + ); }; export const getReplayEvents = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => - d - .select({ question: schema.questions }) - .from(schema.event_queue) - .innerJoin(schema.questions, eq(schema.event_queue.event, EventQueueType.Replay)), - ); + return trycatch(() => + d + .select({ question: schema.questions }) + .from(schema.event_queue) + .innerJoin(schema.questions, eq(schema.event_queue.event, EventQueueType.Replay)), + ); }; export const insertReplayEvents = async ( - questions: Question[], - d: PostgresJsDatabase = db(), + questions: Question[], + d: PostgresJsDatabase = db(), ) => { - const events = questions.map((question) => ({ - event: EventQueueType.Replay, - payload: { question }, - })); - return trycatch(() => d.insert(schema.event_queue).values(events).onConflictDoNothing().returning()); + const events = questions.map((question) => ({ + event: EventQueueType.Replay, + payload: { question }, + })); + return trycatch(() => d.insert(schema.event_queue).values(events).onConflictDoNothing().returning()); }; export const clearReplayEvents = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => - d.delete(schema.event_queue).where(eq(schema.event_queue.event, EventQueueType.Replay)), - ); + return trycatch(() => + d.delete(schema.event_queue).where(eq(schema.event_queue.event, EventQueueType.Replay)), + ); }; export const getAllPrograms = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => - d.selectDistinct({ program: schema.questions.program }).from(schema.questions), - ); + return trycatch(() => + d.selectDistinct({ program: schema.questions.program }).from(schema.questions), + ); }; export const getForumStates = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => d.select().from(schema.forum_state)); + return trycatch(() => d.select().from(schema.forum_state)); }; export const updateForumStates = async ( - states: (typeof schema.forum_state.$inferInsert)[], - d: PostgresJsDatabase = db(), + states: (typeof schema.forum_state.$inferInsert)[], + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => - d - .insert(schema.forum_state) - .values(states) - .onConflictDoUpdate({ - target: schema.forum_state.program, - set: { - open: sql`excluded.open`, - }, - setWhere: sql`${schema.forum_state.open} != excluded.open`, - }), - ); + return trycatch(() => + d + .insert(schema.forum_state) + .values(states) + .onConflictDoUpdate({ + target: schema.forum_state.program, + set: { + open: sql`excluded.open`, + }, + setWhere: sql`${schema.forum_state.open} != excluded.open`, + }), + ); }; diff --git a/packages/store/src/schema_types.ts b/packages/store/src/schema_types.ts index 191705d..b47e5dc 100644 --- a/packages/store/src/schema_types.ts +++ b/packages/store/src/schema_types.ts @@ -1,8 +1,10 @@ import type { Question } from "@qnaplus/scraper"; -import type { forum_state } from "./schema"; +import type { forum_state, metadata } from "./schema"; export type ForumState = typeof forum_state.$inferSelect; +export type Metadata = typeof metadata.$inferSelect; + export type AnsweredPayload = { question: Question; }; diff --git a/packages/utils/src/variables.ts b/packages/utils/src/variables.ts index 6526db1..d89d05c 100644 --- a/packages/utils/src/variables.ts +++ b/packages/utils/src/variables.ts @@ -9,3 +9,9 @@ export const lazy = (init: LazyInitializer) => { return val; }; }; + +export type Nullish = null | undefined; + +export const isNullish = (value: unknown): value is Nullish => { + return value === null || value === undefined; +}; diff --git a/services/updater/src/index.ts b/services/updater/src/index.ts index 043a57b..226e3b8 100644 --- a/services/updater/src/index.ts +++ b/services/updater/src/index.ts @@ -2,19 +2,25 @@ import { getenv, initializeEnv } from "@qnaplus/dotenv"; import { getLoggerInstance } from "@qnaplus/logger"; import type { FetchClient, FetchClientResponse } from "@qnaplus/scraper"; import { CurlImpersonateScrapingClient } from "@qnaplus/scraper-strategies"; -import { testConnection } from "@qnaplus/store"; +import { getMetadata, testConnection } from "@qnaplus/store"; import { Cron } from "croner"; import type { Logger } from "pino"; import { updateDatabase } from "./update_database"; import { updateForumStatus } from "./update_forum_status"; import { updateStorage } from "./update_storage"; +import { isNullish } from "@qnaplus/utils"; const update = async (client: FetchClient, logger: Logger) => { - const { hasUpdates } = await updateDatabase(client, logger); + const [metadataError, meta] = await getMetadata(); + if (metadataError || isNullish(meta)) { + logger?.error({ error: metadataError, meta }, "Error retrieving question metadata, exiting"); + return; + } + const hasUpdates = await updateDatabase(client, meta, logger); if (hasUpdates) { - await updateStorage(logger); + await updateStorage(meta, logger); } - await updateForumStatus(client, logger); + await updateForumStatus(client, meta, logger); }; const start = async (client: FetchClient, logger: Logger) => { diff --git a/services/updater/src/update_database.ts b/services/updater/src/update_database.ts index e58c097..23bd75b 100644 --- a/services/updater/src/update_database.ts +++ b/services/updater/src/update_database.ts @@ -1,57 +1,42 @@ import { - type FetchClient, - type FetchClientResponse, - fetchQuestionsIterative, + type FetchClient, + type FetchClientResponse, + fetchQuestionsIterative, } from "@qnaplus/scraper"; -import { getMetadata, updateQuestions } from "@qnaplus/store"; +import { getMetadata, Metadata, updateQuestions } from "@qnaplus/store"; import type { Logger } from "pino"; export interface DatabaseUpdateStatus { - hasUpdates: boolean; + updateStorage: boolean; } export const updateDatabase = async ( - client: FetchClient, - _logger: Logger, -): Promise => { - const logger = _logger?.child({ label: "update_database" }); - const status: DatabaseUpdateStatus = { - hasUpdates: false, - }; - logger.info("Starting database update."); - const [metadataError, metadata] = await getMetadata(); - if (metadataError) { - logger?.error({ error: metadataError }, "Error retrieving question metadata, exiting"); - return status; - } - if (metadata === undefined) { - logger?.error("Unable to fetch metadata, exiting."); - return status; - } + client: FetchClient, + { start }: Metadata, + _logger: Logger, +): Promise => { + const logger = _logger?.child({ label: "update_database" }); + logger.info("Starting database update."); - const { start } = metadata; + const { questions } = await fetchQuestionsIterative({ + client, + logger, + start, + }); + if (questions.length === 0) { + logger.info("No new questions to insert, returning."); + return false; + } - logger?.info(`Starting update from Q&A ${start}`); - const { questions } = await fetchQuestionsIterative({ - client, - logger, - start, - }); - if (questions.length === 0) { - logger.info("No new questions to insert, returning."); - return status; - } + const [updateError, updates] = await updateQuestions(questions); + if (updateError) { + logger.error( + { error: updateError }, + `Failed to upsert ${questions.length} questions, retrying on next run.`, + ); + return false; + } + logger.info(`${updates.length} new updates detected.`); - const [updateError, updates] = await updateQuestions(questions); - if (updateError) { - logger.error( - { error: updateError }, - `Failed to upsert ${questions.length} questions, retrying on next run.`, - ); - return status; - } - status.hasUpdates = updates.length !== 0; - logger.info(`${updates.length} new updates detected.`); - - return status; + return updates.length !== 0; }; diff --git a/services/updater/src/update_forum_status.ts b/services/updater/src/update_forum_status.ts index 44abee4..a1ac568 100644 --- a/services/updater/src/update_forum_status.ts +++ b/services/updater/src/update_forum_status.ts @@ -1,101 +1,93 @@ import { - type FetchClient, - type FetchClientResponse, - type Season, - checkIfReadOnly, - pingQna, + type FetchClient, + type FetchClientResponse, + type Season, + checkIfReadOnly, + pingQna, } from "@qnaplus/scraper"; -import { getAllPrograms, getForumStates, getMetadata, updateForumStates } from "@qnaplus/store"; +import { getAllPrograms, getForumStates, getMetadata, Metadata, updateForumStates } from "@qnaplus/store"; import type { Logger } from "pino"; type ProgramState = { - program: string; - open: boolean; + program: string; + open: boolean; }; const getNextSeason = (season: Season) => { - if (!/^\d{4}-\d{4}$/.test(season)) { - throw new Error(`Invalid season '${season}' provided.`); - } - const endYear = Number.parseInt(season.split("-")[1]); - return `${endYear}-${endYear + 1}`; + if (!/^\d{4}-\d{4}$/.test(season)) { + throw new Error(`Invalid season '${season}' provided.`); + } + const endYear = Number.parseInt(season.split("-")[1]); + return `${endYear}-${endYear + 1}`; }; export const updateForumStatus = async ( - client: FetchClient, - logger_: Logger, + client: FetchClient, + { currentSeason: season }: Metadata, + logger_: Logger, ) => { - const logger = logger_.child({ label: "update_forum_status" }); - logger.info("Starting programs update."); - const [programsError, programs] = await getAllPrograms(); - if (programsError) { - logger.error( - { error: programsError }, - "An error occurred while trying to aggregate all programs, exiting.", - ); - return; - } - const [metadataError, metadata] = await getMetadata(); - if (metadataError || metadata === undefined) { - logger.error( - { error: metadataError }, - "An error occurred while retrieving metadata, exiting.", - ); - return; - } - const [statesError, states] = await getForumStates(); - if (statesError) { - logger.error( - { error: statesError }, - "An error occurred while trying to fetch existing Q&A states, exiting.", - ); - return; - } - const statesMap = states.reduce>((map, s) => { - map[s.program] = s.open; - return map; - }, {}); + const logger = logger_.child({ label: "update_forum_status" }); + logger.info("Starting programs update."); + const [programsError, programs] = await getAllPrograms(); + if (programsError) { + logger.error( + { error: programsError }, + "An error occurred while trying to aggregate all programs, exiting.", + ); + return; + } + const [statesError, states] = await getForumStates(); + if (statesError) { + logger.error( + { error: statesError }, + "An error occurred while trying to fetch existing Q&A states, exiting.", + ); + return; + } + const statesMap = states.reduce>((map, s) => { + map[s.program] = s.open; + return map; + }, {}); - const season = metadata.currentSeason; - const newStates: ProgramState[] = []; - for (const { program } of programs) { - // get the 'open' state for a given program - // if there is none, default to true so we can initialize one - const currentOpenState = statesMap[program] ?? true; - let open: boolean; - if (currentOpenState || program.toLowerCase() === "judging") { - logger.info(`Forum for ${program} is open, checking readonly status.`); - const readonly = await checkIfReadOnly(program, season, { - client, - logger, - }); - if (readonly === null) { - logger.warn(`Unable to check state for ${program} (${season}), skipping.`); - continue; - } - open = !readonly; - } else { - logger.info( - `Forum for ${program} is closed, checking availability of the next Q&A forum.`, - ); - open = await pingQna(program, getNextSeason(season), { client, logger }); - } + const newStates: ProgramState[] = []; + for (const { program } of programs) { + // get the 'open' state for a given program + // if there is none, default to true so we can initialize one + const currentOpenState = statesMap[program] ?? true; + let open: boolean; + if (currentOpenState || program.toLowerCase() === "judging") { + logger.info(`Forum for ${program} is open, checking readonly status.`); + const readonly = await checkIfReadOnly(program, season, { + client, + logger, + }); + if (readonly === null) { + logger.warn(`Unable to check state for ${program} (${season}), skipping.`); + continue; + } + open = !readonly; + } else { + logger.info( + `Forum for ${program} is closed, checking availability of the next Q&A forum.`, + ); + open = await pingQna(program, getNextSeason(season), { client, logger }); + } - logger.info(`Forum state for ${program}: ${open}`); + logger.info(`Forum state for ${program}: ${open}`); - newStates.push({ program, open }); - } - if (newStates.length === 0) { - logger.warn("Unable to update forum states."); - return; - } - const [updatedError] = await updateForumStates(newStates); - if (updatedError) { - logger.error( - { error: updatedError }, - "An error occurred while updating forum states, exiting.", - ); - return; - } - logger.info("Completed forum states update."); + newStates.push({ program, open }); + } + if (newStates.length === 0) { + logger.warn("Unable to update forum states."); + return; + } + const [updatedError] = await updateForumStates(newStates); + if (updatedError) { + logger.error( + { error: updatedError }, + "An error occurred while updating forum states, exiting.", + ); + return; + } + logger.info("Completed forum states update."); }; diff --git a/services/updater/src/update_storage.ts b/services/updater/src/update_storage.ts index 7bef9aa..fdd7a6f 100644 --- a/services/updater/src/update_storage.ts +++ b/services/updater/src/update_storage.ts @@ -1,56 +1,51 @@ import { getenv } from "@qnaplus/dotenv"; import { Question } from "@qnaplus/scraper"; -import { getAllQuestions, getAllSeasonQuestions, getMetadata, upload } from "@qnaplus/store"; +import { getAllQuestions, getAllSeasonQuestions, Metadata, upload } from "@qnaplus/store"; import { trycatch } from "@qnaplus/utils"; import type { Logger } from "pino"; const update = async (logger: Logger, data: Question[], key: string) => { - const json = JSON.stringify(data); - const buffer = Buffer.from(json, "utf-8"); - const [uploadError] = await trycatch(() => upload(key, buffer, logger)); - if (uploadError) { - logger?.error({ error: uploadError }, "Error while updating storage json"); - return; - } + const json = JSON.stringify(data); + const buffer = Buffer.from(json, "utf-8"); + const [uploadError] = await trycatch(() => upload(key, buffer, logger)); + if (uploadError) { + logger?.error({ error: uploadError }, "Error while updating storage json"); + return; + } }; -export const updateStorage = async (_logger: Logger) => { - const logger = _logger?.child({ label: "update_storage" }); - logger.info("Starting storage update."); +export const updateStorage = async ( + { currentSeason }: Metadata, + _logger: Logger +) => { + const logger = _logger?.child({ label: "update_storage" }); + logger.info("Starting storage update."); - // push all questions - const [questionsError, questions] = await getAllQuestions(); - if (questionsError) { - logger?.error( - { error: questionsError }, - "An error occurred while retreiving all questions from database.", - ); - return; - } - await update(logger, questions, `questions-${getenv("NODE_ENV")}.json`); + // push all questions + const [questionsError, questions] = await getAllQuestions(); + if (questionsError) { + logger?.error( + { error: questionsError }, + "An error occurred while retreiving all questions from database.", + ); + return; + } + await update(logger, questions, `questions-${getenv("NODE_ENV")}.json`); - // push current season questions - const [seasonQuestionsError, seasonQuestions] = await getAllSeasonQuestions(); - if (seasonQuestionsError) { - logger?.error( - { error: seasonQuestionsError }, - "An error occurred while retreiving season questions from database.", - ); - return; - } - const [metaError, meta] = await getMetadata(); - if (metaError || meta === undefined) { - logger?.error( - { error: metaError }, - "An error occurred while retreiving metadata from database.", - ); - return; - } - await update( - logger, - seasonQuestions, - `questions-${getenv("NODE_ENV")}-${meta.currentSeason}.json`, - ); + // push current season questions + const [seasonQuestionsError, seasonQuestions] = await getAllSeasonQuestions(); + if (seasonQuestionsError) { + logger?.error( + { error: seasonQuestionsError }, + "An error occurred while retreiving season questions from database.", + ); + return; + } + await update( + logger, + seasonQuestions, + `questions-${getenv("NODE_ENV")}-${currentSeason}.json`, + ); - logger.info("Completed storage update."); + logger.info("Completed storage update."); }; From a3fa38f8d276ae748cc4cdda164848cd040cfd2f Mon Sep 17 00:00:00 2001 From: Battlesquid Date: Thu, 4 Jun 2026 23:22:17 -0700 Subject: [PATCH 04/10] chore: remove unused imports --- services/updater/src/update_database.ts | 2 +- services/updater/src/update_forum_status.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/services/updater/src/update_database.ts b/services/updater/src/update_database.ts index 23bd75b..2567d48 100644 --- a/services/updater/src/update_database.ts +++ b/services/updater/src/update_database.ts @@ -3,7 +3,7 @@ import { type FetchClientResponse, fetchQuestionsIterative, } from "@qnaplus/scraper"; -import { getMetadata, Metadata, updateQuestions } from "@qnaplus/store"; +import { Metadata, updateQuestions } from "@qnaplus/store"; import type { Logger } from "pino"; export interface DatabaseUpdateStatus { diff --git a/services/updater/src/update_forum_status.ts b/services/updater/src/update_forum_status.ts index a1ac568..da5804a 100644 --- a/services/updater/src/update_forum_status.ts +++ b/services/updater/src/update_forum_status.ts @@ -5,7 +5,7 @@ import { checkIfReadOnly, pingQna, } from "@qnaplus/scraper"; -import { getAllPrograms, getForumStates, getMetadata, Metadata, updateForumStates } from "@qnaplus/store"; +import { getAllPrograms, getForumStates, Metadata, updateForumStates } from "@qnaplus/store"; import type { Logger } from "pino"; type ProgramState = { From 1d5a2eed7e8b58fa889b02c9b4d7587922853386 Mon Sep 17 00:00:00 2001 From: Battlesquid Date: Thu, 4 Jun 2026 23:22:57 -0700 Subject: [PATCH 05/10] chore: drop unused meta query fn --- packages/store/src/database.ts | 4 ---- 1 file changed, 4 deletions(-) diff --git a/packages/store/src/database.ts b/packages/store/src/database.ts index 5ff8b40..5ce97b0 100644 --- a/packages/store/src/database.ts +++ b/packages/store/src/database.ts @@ -160,10 +160,6 @@ export const getMetadata = async (d: PostgresJsDatabase = db()) = return trycatch(() => d.query.metadata.findFirst()); }; -export const getMetadataCurrentSeason = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => d.query.metadata.findFirst({ columns: { currentSeason: true } })); -}; - export const updateMetadata = async ( data: Partial>, d: PostgresJsDatabase = db(), From dc4d864b404da1e86df0c63aa481c19b261cc973 Mon Sep 17 00:00:00 2001 From: Battlesquid <25509915+Battlesquid@users.noreply.github.com> Date: Fri, 5 Jun 2026 10:39:26 -0700 Subject: [PATCH 06/10] refactor: prefer using r2 data to update over db --- packages/store/src/storage.ts | 28 ++++++++++-- packages/utils/src/arrays.ts | 20 ++++++++ services/updater/src/index.ts | 6 +-- services/updater/src/update_database.ts | 9 ++-- services/updater/src/update_storage.ts | 61 ++++++++++++++++--------- 5 files changed, 93 insertions(+), 31 deletions(-) diff --git a/packages/store/src/storage.ts b/packages/store/src/storage.ts index 5558641..13f7081 100644 --- a/packages/store/src/storage.ts +++ b/packages/store/src/storage.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { PutObjectCommand, S3Client } from "@aws-sdk/client-s3"; +import { GetObjectCommand, PutObjectCommand, S3Client } from "@aws-sdk/client-s3"; import { getenv } from "@qnaplus/dotenv"; import { lazy, trycatch } from "@qnaplus/utils"; import type { Logger } from "pino"; @@ -20,7 +20,7 @@ const r2 = lazy( }), ); -export const upload = async (key: string, buffer: Buffer, logger: Logger) => { +export const uploadObject = async (key: string, buffer: Buffer, logger: Logger) => { const command = new PutObjectCommand({ Key: key, Bucket: BUCKET, @@ -31,7 +31,29 @@ export const upload = async (key: string, buffer: Buffer, logger: Logger) => { }); const [error] = await trycatch(() => r2().send(command)); if (error) { - logger.error({ error }, "An error occurred while uploading questions to bucket."); + logger.error({ error }, "An error occurred while uploading data to bucket."); return; } }; + +export const getObject = async (key: string, logger: Logger): Promise => { + const command = new GetObjectCommand({ + Key: key, + Bucket: BUCKET + }); + const [responseError, response] = await trycatch(() => r2().send(command)); + if (responseError) { + logger.error({ error: responseError }, "An error occurred while getting object from bucket."); + return null; + } + if (response.Body === undefined) { + logger.error("Body is not present on response."); + return null + } + const [dataError, data] = await trycatch(() => response.Body!.transformToString()); + if (dataError) { + logger.error({ error: dataError }, "An error occurred while reading the response body."); + return null; + } + return data; +} diff --git a/packages/utils/src/arrays.ts b/packages/utils/src/arrays.ts index 1c862c6..d4fb586 100644 --- a/packages/utils/src/arrays.ts +++ b/packages/utils/src/arrays.ts @@ -29,3 +29,23 @@ export const entries = (obj: T): Entries => { // biome-ignore lint/suspicious/noExplicitAny: return Object.entries(obj) as any; }; + +export const mergeByKey = < + K extends string, + A extends Record, + B extends Record, +>( + key: K, + a: A[], + b: B[], +): (A | B | (A & B))[] => { + const map = new Map(); + for (const item of a) { + map.set(item[key], item); + } + for (const item of b) { + const existing = map.get(item[key]); + map.set(item[key], existing ? { ...existing, ...item } : item); + } + return [...map.values()]; +}; diff --git a/services/updater/src/index.ts b/services/updater/src/index.ts index 226e3b8..19d040f 100644 --- a/services/updater/src/index.ts +++ b/services/updater/src/index.ts @@ -16,9 +16,9 @@ const update = async (client: FetchClient, logger: Logger) logger?.error({ error: metadataError, meta }, "Error retrieving question metadata, exiting"); return; } - const hasUpdates = await updateDatabase(client, meta, logger); - if (hasUpdates) { - await updateStorage(meta, logger); + const updates = await updateDatabase(client, meta, logger); + if (updates.length > 0) { + await updateStorage(updates, meta, logger); } await updateForumStatus(client, meta, logger); }; diff --git a/services/updater/src/update_database.ts b/services/updater/src/update_database.ts index 2567d48..34de433 100644 --- a/services/updater/src/update_database.ts +++ b/services/updater/src/update_database.ts @@ -2,6 +2,7 @@ import { type FetchClient, type FetchClientResponse, fetchQuestionsIterative, + Question, } from "@qnaplus/scraper"; import { Metadata, updateQuestions } from "@qnaplus/store"; import type { Logger } from "pino"; @@ -14,7 +15,7 @@ export const updateDatabase = async ( client: FetchClient, { start }: Metadata, _logger: Logger, -): Promise => { +): Promise => { const logger = _logger?.child({ label: "update_database" }); logger.info("Starting database update."); @@ -25,7 +26,7 @@ export const updateDatabase = async ( }); if (questions.length === 0) { logger.info("No new questions to insert, returning."); - return false; + return []; } const [updateError, updates] = await updateQuestions(questions); @@ -34,9 +35,9 @@ export const updateDatabase = async ( { error: updateError }, `Failed to upsert ${questions.length} questions, retrying on next run.`, ); - return false; + return []; } logger.info(`${updates.length} new updates detected.`); - return updates.length !== 0; + return updates; }; diff --git a/services/updater/src/update_storage.ts b/services/updater/src/update_storage.ts index fdd7a6f..53486bf 100644 --- a/services/updater/src/update_storage.ts +++ b/services/updater/src/update_storage.ts @@ -1,50 +1,69 @@ import { getenv } from "@qnaplus/dotenv"; import { Question } from "@qnaplus/scraper"; -import { getAllQuestions, getAllSeasonQuestions, Metadata, upload } from "@qnaplus/store"; -import { trycatch } from "@qnaplus/utils"; +import { getObject, Metadata, uploadObject } from "@qnaplus/store"; +import { mergeByKey, trycatch } from "@qnaplus/utils"; import type { Logger } from "pino"; const update = async (logger: Logger, data: Question[], key: string) => { const json = JSON.stringify(data); const buffer = Buffer.from(json, "utf-8"); - const [uploadError] = await trycatch(() => upload(key, buffer, logger)); + const [uploadError] = await trycatch(() => uploadObject(key, buffer, logger)); if (uploadError) { logger?.error({ error: uploadError }, "Error while updating storage json"); return; } }; +const get = async (key: string, logger: Logger): Promise => { + const obj = await getObject(key, logger); + if (obj === null) { + logger.warn( + `Unable to retreive questions data at '${key}' from storage.`, + ); + return null; + } + const [parseError, parsedData] = await trycatch((): Promise => JSON.parse(obj)); + if (parseError) { + logger.error( + { error: parseError }, + "Unable to parse questions from storage.", + ); + return null; + } + return parsedData; +} + +const ALL_QUESTIONS_KEY = `questions-${getenv("NODE_ENV")}.json`; + export const updateStorage = async ( + updates: Question[], { currentSeason }: Metadata, _logger: Logger ) => { - const logger = _logger?.child({ label: "update_storage" }); + const logger = _logger.child({ label: "update_storage" }); logger.info("Starting storage update."); - // push all questions - const [questionsError, questions] = await getAllQuestions(); - if (questionsError) { - logger?.error( - { error: questionsError }, - "An error occurred while retreiving all questions from database.", - ); + const SEASON_QUESTIONS_KEY = `questions-${getenv("NODE_ENV")}-${currentSeason}.json`; + const seasonQuestions = await get(SEASON_QUESTIONS_KEY, logger); + if (seasonQuestions === null) { return; } - await update(logger, questions, `questions-${getenv("NODE_ENV")}.json`); + const updatedSeasonQuestions = mergeByKey("id", updates, seasonQuestions); + await update( + logger, + updatedSeasonQuestions, + SEASON_QUESTIONS_KEY, + ); - // push current season questions - const [seasonQuestionsError, seasonQuestions] = await getAllSeasonQuestions(); - if (seasonQuestionsError) { - logger?.error( - { error: seasonQuestionsError }, - "An error occurred while retreiving season questions from database.", - ); + const questions = await get(ALL_QUESTIONS_KEY, logger); + if (questions === null) { return; } + const updatedQuestions = mergeByKey("id", updatedSeasonQuestions, questions); await update( logger, - seasonQuestions, - `questions-${getenv("NODE_ENV")}-${currentSeason}.json`, + updatedQuestions, + ALL_QUESTIONS_KEY ); logger.info("Completed storage update."); From 3862fe9b80113c35930297428a490392e1a5d6fd Mon Sep 17 00:00:00 2001 From: Battlesquid <25509915+Battlesquid@users.noreply.github.com> Date: Fri, 5 Jun 2026 21:37:46 -0700 Subject: [PATCH 07/10] fix: put key inside fn to allow env vars to load first --- services/updater/src/update_storage.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/services/updater/src/update_storage.ts b/services/updater/src/update_storage.ts index 53486bf..53059c9 100644 --- a/services/updater/src/update_storage.ts +++ b/services/updater/src/update_storage.ts @@ -33,8 +33,6 @@ const get = async (key: string, logger: Logger): Promise => { return parsedData; } -const ALL_QUESTIONS_KEY = `questions-${getenv("NODE_ENV")}.json`; - export const updateStorage = async ( updates: Question[], { currentSeason }: Metadata, @@ -43,7 +41,9 @@ export const updateStorage = async ( const logger = _logger.child({ label: "update_storage" }); logger.info("Starting storage update."); + const ALL_QUESTIONS_KEY = `questions-${getenv("NODE_ENV")}.json`; const SEASON_QUESTIONS_KEY = `questions-${getenv("NODE_ENV")}-${currentSeason}.json`; + const seasonQuestions = await get(SEASON_QUESTIONS_KEY, logger); if (seasonQuestions === null) { return; From 3df69ec3686f0d4e37a8f320e4f428163628007e Mon Sep 17 00:00:00 2001 From: Battlesquid <25509915+Battlesquid@users.noreply.github.com> Date: Sat, 6 Jun 2026 15:19:08 -0700 Subject: [PATCH 08/10] chore: bump ver --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index 2d64d54..bf0f4b1 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "qnaplus", - "version": "1.1.0", + "version": "1.2.0", "description": "", "private": true, "author": "battlesqui_d", From d5eeb40643fddc6ebf24b5d7bce21d82bd152e82 Mon Sep 17 00:00:00 2001 From: Battlesquid <25509915+Battlesquid@users.noreply.github.com> Date: Sat, 6 Jun 2026 15:20:57 -0700 Subject: [PATCH 09/10] fix: sort questions by id before upload --- services/updater/src/update_storage.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/services/updater/src/update_storage.ts b/services/updater/src/update_storage.ts index 53059c9..e0bf2ab 100644 --- a/services/updater/src/update_storage.ts +++ b/services/updater/src/update_storage.ts @@ -48,7 +48,8 @@ export const updateStorage = async ( if (seasonQuestions === null) { return; } - const updatedSeasonQuestions = mergeByKey("id", updates, seasonQuestions); + const updatedSeasonQuestions = mergeByKey("id", updates, seasonQuestions) + .toSorted((a, b) => Number.parseInt(b.id) - Number.parseInt(a.id)); await update( logger, updatedSeasonQuestions, @@ -59,7 +60,8 @@ export const updateStorage = async ( if (questions === null) { return; } - const updatedQuestions = mergeByKey("id", updatedSeasonQuestions, questions); + const updatedQuestions = mergeByKey("id", updatedSeasonQuestions, questions) + .toSorted((a, b) => Number.parseInt(b.id) - Number.parseInt(a.id)); await update( logger, updatedQuestions, From a45ea69b75f7fef1ae22d7ce705f7a3076dcf5af Mon Sep 17 00:00:00 2001 From: Battlesquid <25509915+Battlesquid@users.noreply.github.com> Date: Sat, 6 Jun 2026 22:24:48 -0700 Subject: [PATCH 10/10] chore: neat --- packages/store/src/database.ts | 294 ++++++++++---------- packages/store/src/storage.ts | 11 +- services/bot/src/broadcaster.ts | 8 +- services/updater/src/index.ts | 61 ++-- services/updater/src/update_database.ts | 58 ++-- services/updater/src/update_forum_status.ts | 152 +++++----- services/updater/src/update_storage.ts | 97 +++---- 7 files changed, 342 insertions(+), 339 deletions(-) diff --git a/packages/store/src/database.ts b/packages/store/src/database.ts index 5ce97b0..2043351 100644 --- a/packages/store/src/database.ts +++ b/packages/store/src/database.ts @@ -11,217 +11,219 @@ const pg = lazy(() => postgres(getenv("SUPABASE_TRANSACTION_URL"))); export const db = lazy(() => drizzle({ schema, client: pg() })); export const disconnectPgClient = async () => { - const client = pg(); - if (client !== null) { - await client.end(); - } + const client = pg(); + if (client !== null) { + await client.end(); + } }; export const METADATA_ROW_ID = 0; export const testConnection = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => d.execute(sql`select 1`)); + return trycatch(() => d.execute(sql`select 1`)); }; export const getQuestion = async ( - id: Question["id"], - d: PostgresJsDatabase = db(), + id: Question["id"], + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => d.query.questions.findFirst({ where: eq(schema.questions.id, id) })); + return trycatch(() => d.query.questions.findFirst({ where: eq(schema.questions.id, id) })); }; export const getAllQuestions = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => d.select().from(schema.questions)); + return trycatch(() => d.select().from(schema.questions)); }; export const getAllSeasonQuestions = async (d: PostgresJsDatabase = db()) => { - const currentSeason = d.$with("current_season").as( - d - .select({ season: sql`${schema.metadata.currentSeason}`.as("s") }) - .from(schema.metadata) - .limit(1), - ); - return trycatch(() => - d - .with(currentSeason) - .select(getTableColumns(schema.questions)) - .from(schema.questions) - .innerJoin(currentSeason, eq(schema.questions.season, currentSeason.season)), - ); + const currentSeason = d.$with("current_season").as( + d + .select({ season: sql`${schema.metadata.currentSeason}`.as("s") }) + .from(schema.metadata) + .limit(1), + ); + return trycatch(() => + d + .with(currentSeason) + .select(getTableColumns(schema.questions)) + .from(schema.questions) + .innerJoin(currentSeason, eq(schema.questions.season, currentSeason.season)), + ); }; export const getAnsweredQuestionsNewerThanDate = async ( - ms: number, - d: PostgresJsDatabase = db(), + ms: number, + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => - d - .select() - .from(schema.questions) - .where( - and( - gte(schema.questions.answeredTimestampMs, ms), - eq(schema.questions.answered, true), - ), - ), - ); + return trycatch(() => + d + .select() + .from(schema.questions) + .where( + and( + gte(schema.questions.answeredTimestampMs, ms), + eq(schema.questions.answered, true), + ), + ), + ); }; export const insertQuestions = async ( - data: Question[], - d: PostgresJsDatabase = db(), + data: Question[], + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => d.insert(schema.questions).values(data)); + return trycatch(() => d.insert(schema.questions).values(data)); }; const EXCLUDED_QUESTION = { - id: sql`excluded.id`, - url: sql`excluded.url`, - author: sql`excluded.author`, - program: sql`excluded.program`, - title: sql`excluded.title`, - question: sql`excluded.question`, - questionRaw: sql`excluded."questionRaw"`, - answer: sql`excluded.answer`, - answerRaw: sql`excluded."answerRaw"`, - season: sql`excluded.season`, - askedTimestamp: sql`excluded."askedTimestamp"`, - askedTimestampMs: sql`excluded."askedTimestampMs"`, - answeredTimestamp: sql`excluded."answeredTimestamp"`, - answeredTimestampMs: sql`excluded."answeredTimestampMs"`, - answered: sql`excluded.answered`, - tags: sql`excluded.tags`, + id: sql`excluded.id`, + url: sql`excluded.url`, + author: sql`excluded.author`, + program: sql`excluded.program`, + title: sql`excluded.title`, + question: sql`excluded.question`, + questionRaw: sql`excluded."questionRaw"`, + answer: sql`excluded.answer`, + answerRaw: sql`excluded."answerRaw"`, + season: sql`excluded.season`, + askedTimestamp: sql`excluded."askedTimestamp"`, + askedTimestampMs: sql`excluded."askedTimestampMs"`, + answeredTimestamp: sql`excluded."answeredTimestamp"`, + answeredTimestampMs: sql`excluded."answeredTimestampMs"`, + answered: sql`excluded.answered`, + tags: sql`excluded.tags`, }; const QUESTION_UPDATED_QUERY = or( - sql`${schema.questions.question} != excluded.question`, - sql`${schema.questions.answer} IS DISTINCT FROM excluded.answer`, - sql`${schema.questions.answered} != excluded.answered`, + sql`${schema.questions.question} != excluded.question`, + sql`${schema.questions.answer} IS DISTINCT FROM excluded.answer`, + sql`${schema.questions.answered} != excluded.answered`, ); export const updateQuestions = async ( - data: Question[], - d: PostgresJsDatabase = db(), + data: Question[], + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => - d - .insert(schema.questions) - .values(data) - .onConflictDoUpdate({ - target: schema.questions.id, - set: EXCLUDED_QUESTION, - setWhere: QUESTION_UPDATED_QUERY, - }) - .returning(), - ); + return trycatch(() => + d + .insert(schema.questions) + .values(data) + .onConflictDoUpdate({ + target: schema.questions.id, + set: EXCLUDED_QUESTION, + setWhere: QUESTION_UPDATED_QUERY, + }) + .returning(), + ); }; export const getEventQueue = async (d: PostgresJsDatabase = db()) => { - const formattedPayloads = d.$with("formatted_payloads").as( - d - .select({ - event: schema.event_queue.event, - payload: - sql`jsonb_build_object('id', ${schema.event_queue.id}, 'payload', ${schema.event_queue.payload})`.as( - "payload", - ), - }) - .from(schema.event_queue) - .orderBy(schema.event_queue.timestamp), - ); - - const aggregatedPayloads = d.$with("aggregated_payloads").as( - d - .select({ - event: formattedPayloads.event, - payloads: sql`array_agg(${formattedPayloads.payload})`.as("payloads"), - }) - .from(formattedPayloads) - .groupBy(formattedPayloads.event), - ); - return trycatch(() => - d - .with(formattedPayloads, aggregatedPayloads) - .select({ - queue: sql`jsonb_object_agg(${schema.event_queue.event}, ${aggregatedPayloads.payloads})`, - }) - .from(aggregatedPayloads), - ); + const formattedPayloads = d.$with("formatted_payloads").as( + d + .select({ + event: schema.event_queue.event, + payload: + sql`jsonb_build_object('id', ${schema.event_queue.id}, 'payload', ${schema.event_queue.payload})`.as( + "payload", + ), + }) + .from(schema.event_queue) + .orderBy(schema.event_queue.timestamp), + ); + + const aggregatedPayloads = d.$with("aggregated_payloads").as( + d + .select({ + event: formattedPayloads.event, + payloads: sql`array_agg(${formattedPayloads.payload})`.as("payloads"), + }) + .from(formattedPayloads) + .groupBy(formattedPayloads.event), + ); + return trycatch(() => + d + .with(formattedPayloads, aggregatedPayloads) + .select({ + queue: sql`jsonb_object_agg(${schema.event_queue.event}, ${aggregatedPayloads.payloads})`, + }) + .from(aggregatedPayloads), + ); }; export const clearEventQueue = async ( - ids: string[], - d: PostgresJsDatabase = db(), + ids: string[], + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => d.delete(schema.event_queue).where(inArray(schema.event_queue.id, ids))); + return trycatch(() => d.delete(schema.event_queue).where(inArray(schema.event_queue.id, ids))); }; export const getMetadata = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => d.query.metadata.findFirst()); + return trycatch(() => d.query.metadata.findFirst()); }; export const updateMetadata = async ( - data: Partial>, - d: PostgresJsDatabase = db(), + data: Partial>, + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => - d - .update(schema.metadata) - .set({ ...data }) - .where(eq(schema.metadata.id, METADATA_ROW_ID)), - ); + return trycatch(() => + d + .update(schema.metadata) + .set({ ...data }) + .where(eq(schema.metadata.id, METADATA_ROW_ID)), + ); }; export const getReplayEvents = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => - d - .select({ question: schema.questions }) - .from(schema.event_queue) - .innerJoin(schema.questions, eq(schema.event_queue.event, EventQueueType.Replay)), - ); + return trycatch(() => + d + .select({ question: schema.questions }) + .from(schema.event_queue) + .innerJoin(schema.questions, eq(schema.event_queue.event, EventQueueType.Replay)), + ); }; export const insertReplayEvents = async ( - questions: Question[], - d: PostgresJsDatabase = db(), + questions: Question[], + d: PostgresJsDatabase = db(), ) => { - const events = questions.map((question) => ({ - event: EventQueueType.Replay, - payload: { question }, - })); - return trycatch(() => d.insert(schema.event_queue).values(events).onConflictDoNothing().returning()); + const events = questions.map((question) => ({ + event: EventQueueType.Replay, + payload: { question }, + })); + return trycatch(() => + d.insert(schema.event_queue).values(events).onConflictDoNothing().returning(), + ); }; export const clearReplayEvents = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => - d.delete(schema.event_queue).where(eq(schema.event_queue.event, EventQueueType.Replay)), - ); + return trycatch(() => + d.delete(schema.event_queue).where(eq(schema.event_queue.event, EventQueueType.Replay)), + ); }; export const getAllPrograms = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => - d.selectDistinct({ program: schema.questions.program }).from(schema.questions), - ); + return trycatch(() => + d.selectDistinct({ program: schema.questions.program }).from(schema.questions), + ); }; export const getForumStates = async (d: PostgresJsDatabase = db()) => { - return trycatch(() => d.select().from(schema.forum_state)); + return trycatch(() => d.select().from(schema.forum_state)); }; export const updateForumStates = async ( - states: (typeof schema.forum_state.$inferInsert)[], - d: PostgresJsDatabase = db(), + states: (typeof schema.forum_state.$inferInsert)[], + d: PostgresJsDatabase = db(), ) => { - return trycatch(() => - d - .insert(schema.forum_state) - .values(states) - .onConflictDoUpdate({ - target: schema.forum_state.program, - set: { - open: sql`excluded.open`, - }, - setWhere: sql`${schema.forum_state.open} != excluded.open`, - }), - ); + return trycatch(() => + d + .insert(schema.forum_state) + .values(states) + .onConflictDoUpdate({ + target: schema.forum_state.program, + set: { + open: sql`excluded.open`, + }, + setWhere: sql`${schema.forum_state.open} != excluded.open`, + }), + ); }; diff --git a/packages/store/src/storage.ts b/packages/store/src/storage.ts index 13f7081..4f61ba8 100644 --- a/packages/store/src/storage.ts +++ b/packages/store/src/storage.ts @@ -39,16 +39,19 @@ export const uploadObject = async (key: string, buffer: Buffer, logger: Logger) export const getObject = async (key: string, logger: Logger): Promise => { const command = new GetObjectCommand({ Key: key, - Bucket: BUCKET + Bucket: BUCKET, }); const [responseError, response] = await trycatch(() => r2().send(command)); if (responseError) { - logger.error({ error: responseError }, "An error occurred while getting object from bucket."); + logger.error( + { error: responseError }, + "An error occurred while getting object from bucket.", + ); return null; } if (response.Body === undefined) { logger.error("Body is not present on response."); - return null + return null; } const [dataError, data] = await trycatch(() => response.Body!.transformToString()); if (dataError) { @@ -56,4 +59,4 @@ export const getObject = async (key: string, logger: Logger): Promise, logger: Logger) => { - const [metadataError, meta] = await getMetadata(); - if (metadataError || isNullish(meta)) { - logger?.error({ error: metadataError, meta }, "Error retrieving question metadata, exiting"); - return; - } - const updates = await updateDatabase(client, meta, logger); - if (updates.length > 0) { - await updateStorage(updates, meta, logger); - } - await updateForumStatus(client, meta, logger); + const [metadataError, meta] = await getMetadata(); + if (metadataError || isNullish(meta)) { + logger?.error( + { error: metadataError, meta }, + "Error retrieving question metadata, exiting", + ); + return; + } + const updates = await updateDatabase(client, meta, logger); + if (updates.length > 0) { + await updateStorage(updates, meta, logger); + } + await updateForumStatus(client, meta, logger); }; const start = async (client: FetchClient, logger: Logger) => { - logger.info("Starting database update job"); - const job = new Cron(getenv("DATABASE_UPDATE_INTERVAL"), () => update(client, logger), { - name: "updater", - protect: true, - catch(error) { - logger.error({ error }, "An error occurred while updating database."); - }, - }); - job.trigger(); + logger.info("Starting database update job"); + const job = new Cron(getenv("DATABASE_UPDATE_INTERVAL"), () => update(client, logger), { + name: "updater", + protect: true, + catch(error) { + logger.error({ error }, "An error occurred while updating database."); + }, + }); + job.trigger(); }; (async () => { - await initializeEnv(); + await initializeEnv(); - const logger = getLoggerInstance("qnaplus-updater"); - logger.info("Starting updater service"); + const logger = getLoggerInstance("qnaplus-updater"); + logger.info("Starting updater service"); - const [error] = await testConnection(); - if (error) { - logger.error({ error }, "Unable to establish database connection, exiting."); - process.exit(1); - } + const [error] = await testConnection(); + if (error) { + logger.error({ error }, "Unable to establish database connection, exiting."); + process.exit(1); + } - const client = new CurlImpersonateScrapingClient(logger); + const client = new CurlImpersonateScrapingClient(logger); - start(client, logger); + start(client, logger); })(); diff --git a/services/updater/src/update_database.ts b/services/updater/src/update_database.ts index 34de433..eff275a 100644 --- a/services/updater/src/update_database.ts +++ b/services/updater/src/update_database.ts @@ -1,43 +1,43 @@ import { - type FetchClient, - type FetchClientResponse, - fetchQuestionsIterative, - Question, + type FetchClient, + type FetchClientResponse, + fetchQuestionsIterative, + Question, } from "@qnaplus/scraper"; import { Metadata, updateQuestions } from "@qnaplus/store"; import type { Logger } from "pino"; export interface DatabaseUpdateStatus { - updateStorage: boolean; + updateStorage: boolean; } export const updateDatabase = async ( - client: FetchClient, - { start }: Metadata, - _logger: Logger, + client: FetchClient, + { start }: Metadata, + _logger: Logger, ): Promise => { - const logger = _logger?.child({ label: "update_database" }); - logger.info("Starting database update."); + const logger = _logger?.child({ label: "update_database" }); + logger.info("Starting database update."); - const { questions } = await fetchQuestionsIterative({ - client, - logger, - start, - }); - if (questions.length === 0) { - logger.info("No new questions to insert, returning."); - return []; - } + const { questions } = await fetchQuestionsIterative({ + client, + logger, + start, + }); + if (questions.length === 0) { + logger.info("No new questions to insert, returning."); + return []; + } - const [updateError, updates] = await updateQuestions(questions); - if (updateError) { - logger.error( - { error: updateError }, - `Failed to upsert ${questions.length} questions, retrying on next run.`, - ); - return []; - } - logger.info(`${updates.length} new updates detected.`); + const [updateError, updates] = await updateQuestions(questions); + if (updateError) { + logger.error( + { error: updateError }, + `Failed to upsert ${questions.length} questions, retrying on next run.`, + ); + return []; + } + logger.info(`${updates.length} new updates detected.`); - return updates; + return updates; }; diff --git a/services/updater/src/update_forum_status.ts b/services/updater/src/update_forum_status.ts index da5804a..0853599 100644 --- a/services/updater/src/update_forum_status.ts +++ b/services/updater/src/update_forum_status.ts @@ -1,93 +1,93 @@ import { - type FetchClient, - type FetchClientResponse, - type Season, - checkIfReadOnly, - pingQna, + type FetchClient, + type FetchClientResponse, + type Season, + checkIfReadOnly, + pingQna, } from "@qnaplus/scraper"; import { getAllPrograms, getForumStates, Metadata, updateForumStates } from "@qnaplus/store"; import type { Logger } from "pino"; type ProgramState = { - program: string; - open: boolean; + program: string; + open: boolean; }; const getNextSeason = (season: Season) => { - if (!/^\d{4}-\d{4}$/.test(season)) { - throw new Error(`Invalid season '${season}' provided.`); - } - const endYear = Number.parseInt(season.split("-")[1]); - return `${endYear}-${endYear + 1}`; + if (!/^\d{4}-\d{4}$/.test(season)) { + throw new Error(`Invalid season '${season}' provided.`); + } + const endYear = Number.parseInt(season.split("-")[1]); + return `${endYear}-${endYear + 1}`; }; export const updateForumStatus = async ( - client: FetchClient, - { currentSeason: season }: Metadata, - logger_: Logger, + client: FetchClient, + { currentSeason: season }: Metadata, + logger_: Logger, ) => { - const logger = logger_.child({ label: "update_forum_status" }); - logger.info("Starting programs update."); - const [programsError, programs] = await getAllPrograms(); - if (programsError) { - logger.error( - { error: programsError }, - "An error occurred while trying to aggregate all programs, exiting.", - ); - return; - } - const [statesError, states] = await getForumStates(); - if (statesError) { - logger.error( - { error: statesError }, - "An error occurred while trying to fetch existing Q&A states, exiting.", - ); - return; - } - const statesMap = states.reduce>((map, s) => { - map[s.program] = s.open; - return map; - }, {}); + const logger = logger_.child({ label: "update_forum_status" }); + logger.info("Starting programs update."); + const [programsError, programs] = await getAllPrograms(); + if (programsError) { + logger.error( + { error: programsError }, + "An error occurred while trying to aggregate all programs, exiting.", + ); + return; + } + const [statesError, states] = await getForumStates(); + if (statesError) { + logger.error( + { error: statesError }, + "An error occurred while trying to fetch existing Q&A states, exiting.", + ); + return; + } + const statesMap = states.reduce>((map, s) => { + map[s.program] = s.open; + return map; + }, {}); - const newStates: ProgramState[] = []; - for (const { program } of programs) { - // get the 'open' state for a given program - // if there is none, default to true so we can initialize one - const currentOpenState = statesMap[program] ?? true; - let open: boolean; - if (currentOpenState || program.toLowerCase() === "judging") { - logger.info(`Forum for ${program} is open, checking readonly status.`); - const readonly = await checkIfReadOnly(program, season, { - client, - logger, - }); - if (readonly === null) { - logger.warn(`Unable to check state for ${program} (${season}), skipping.`); - continue; - } - open = !readonly; - } else { - logger.info( - `Forum for ${program} is closed, checking availability of the next Q&A forum.`, - ); - open = await pingQna(program, getNextSeason(season), { client, logger }); - } + const newStates: ProgramState[] = []; + for (const { program } of programs) { + // get the 'open' state for a given program + // if there is none, default to true so we can initialize one + const currentOpenState = statesMap[program] ?? true; + let open: boolean; + if (currentOpenState || program.toLowerCase() === "judging") { + logger.info(`Forum for ${program} is open, checking readonly status.`); + const readonly = await checkIfReadOnly(program, season, { + client, + logger, + }); + if (readonly === null) { + logger.warn(`Unable to check state for ${program} (${season}), skipping.`); + continue; + } + open = !readonly; + } else { + logger.info( + `Forum for ${program} is closed, checking availability of the next Q&A forum.`, + ); + open = await pingQna(program, getNextSeason(season), { client, logger }); + } - logger.info(`Forum state for ${program}: ${open}`); + logger.info(`Forum state for ${program}: ${open}`); - newStates.push({ program, open }); - } - if (newStates.length === 0) { - logger.warn("Unable to update forum states."); - return; - } - const [updatedError] = await updateForumStates(newStates); - if (updatedError) { - logger.error( - { error: updatedError }, - "An error occurred while updating forum states, exiting.", - ); - return; - } - logger.info("Completed forum states update."); + newStates.push({ program, open }); + } + if (newStates.length === 0) { + logger.warn("Unable to update forum states."); + return; + } + const [updatedError] = await updateForumStates(newStates); + if (updatedError) { + logger.error( + { error: updatedError }, + "An error occurred while updating forum states, exiting.", + ); + return; + } + logger.info("Completed forum states update."); }; diff --git a/services/updater/src/update_storage.ts b/services/updater/src/update_storage.ts index e0bf2ab..5360848 100644 --- a/services/updater/src/update_storage.ts +++ b/services/updater/src/update_storage.ts @@ -5,68 +5,57 @@ import { mergeByKey, trycatch } from "@qnaplus/utils"; import type { Logger } from "pino"; const update = async (logger: Logger, data: Question[], key: string) => { - const json = JSON.stringify(data); - const buffer = Buffer.from(json, "utf-8"); - const [uploadError] = await trycatch(() => uploadObject(key, buffer, logger)); - if (uploadError) { - logger?.error({ error: uploadError }, "Error while updating storage json"); - return; - } + const json = JSON.stringify(data); + const buffer = Buffer.from(json, "utf-8"); + const [uploadError] = await trycatch(() => uploadObject(key, buffer, logger)); + if (uploadError) { + logger?.error({ error: uploadError }, "Error while updating storage json"); + return; + } }; const get = async (key: string, logger: Logger): Promise => { - const obj = await getObject(key, logger); - if (obj === null) { - logger.warn( - `Unable to retreive questions data at '${key}' from storage.`, - ); - return null; - } - const [parseError, parsedData] = await trycatch((): Promise => JSON.parse(obj)); - if (parseError) { - logger.error( - { error: parseError }, - "Unable to parse questions from storage.", - ); - return null; - } - return parsedData; -} + const obj = await getObject(key, logger); + if (obj === null) { + logger.warn(`Unable to retreive questions data at '${key}' from storage.`); + return null; + } + const [parseError, parsedData] = await trycatch((): Promise => JSON.parse(obj)); + if (parseError) { + logger.error({ error: parseError }, "Unable to parse questions from storage."); + return null; + } + return parsedData; +}; export const updateStorage = async ( - updates: Question[], - { currentSeason }: Metadata, - _logger: Logger + updates: Question[], + { currentSeason }: Metadata, + _logger: Logger, ) => { - const logger = _logger.child({ label: "update_storage" }); - logger.info("Starting storage update."); + const logger = _logger.child({ label: "update_storage" }); + logger.info("Starting storage update."); - const ALL_QUESTIONS_KEY = `questions-${getenv("NODE_ENV")}.json`; - const SEASON_QUESTIONS_KEY = `questions-${getenv("NODE_ENV")}-${currentSeason}.json`; + const ALL_QUESTIONS_KEY = `questions-${getenv("NODE_ENV")}.json`; + const SEASON_QUESTIONS_KEY = `questions-${getenv("NODE_ENV")}-${currentSeason}.json`; - const seasonQuestions = await get(SEASON_QUESTIONS_KEY, logger); - if (seasonQuestions === null) { - return; - } - const updatedSeasonQuestions = mergeByKey("id", updates, seasonQuestions) - .toSorted((a, b) => Number.parseInt(b.id) - Number.parseInt(a.id)); - await update( - logger, - updatedSeasonQuestions, - SEASON_QUESTIONS_KEY, - ); + const seasonQuestions = await get(SEASON_QUESTIONS_KEY, logger); + if (seasonQuestions === null) { + return; + } + const updatedSeasonQuestions = mergeByKey("id", updates, seasonQuestions).toSorted( + (a, b) => Number.parseInt(b.id) - Number.parseInt(a.id), + ); + await update(logger, updatedSeasonQuestions, SEASON_QUESTIONS_KEY); - const questions = await get(ALL_QUESTIONS_KEY, logger); - if (questions === null) { - return; - } - const updatedQuestions = mergeByKey("id", updatedSeasonQuestions, questions) - .toSorted((a, b) => Number.parseInt(b.id) - Number.parseInt(a.id)); - await update( - logger, - updatedQuestions, - ALL_QUESTIONS_KEY - ); + const questions = await get(ALL_QUESTIONS_KEY, logger); + if (questions === null) { + return; + } + const updatedQuestions = mergeByKey("id", updatedSeasonQuestions, questions).toSorted( + (a, b) => Number.parseInt(b.id) - Number.parseInt(a.id), + ); + await update(logger, updatedQuestions, ALL_QUESTIONS_KEY); - logger.info("Completed storage update."); + logger.info("Completed storage update."); };