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", diff --git a/packages/store/src/database.ts b/packages/store/src/database.ts index ab1f4bb..2043351 100644 --- a/packages/store/src/database.ts +++ b/packages/store/src/database.ts @@ -189,7 +189,9 @@ 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/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/store/src/storage.ts b/packages/store/src/storage.ts index 5558641..4f61ba8 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,32 @@ 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/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/bot/src/broadcaster.ts b/services/bot/src/broadcaster.ts index f2bd07b..8a25671 100644 --- a/services/bot/src/broadcaster.ts +++ b/services/bot/src/broadcaster.ts @@ -8,7 +8,13 @@ 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 +45,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(); } }; diff --git a/services/updater/src/index.ts b/services/updater/src/index.ts index 1f423c5..3c39857 100644 --- a/services/updater/src/index.ts +++ b/services/updater/src/index.ts @@ -2,17 +2,28 @@ 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) => { - await updateDatabase(client, logger); - await updateStorage(logger); - await updateForumStatus(client, 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) => { diff --git a/services/updater/src/update_database.ts b/services/updater/src/update_database.ts index f57d58c..eff275a 100644 --- a/services/updater/src/update_database.ts +++ b/services/updater/src/update_database.ts @@ -2,8 +2,9 @@ import { type FetchClient, type FetchClientResponse, fetchQuestionsIterative, + Question, } from "@qnaplus/scraper"; -import { getMetadata, updateQuestions } from "@qnaplus/store"; +import { Metadata, updateQuestions } from "@qnaplus/store"; import type { Logger } from "pino"; export interface DatabaseUpdateStatus { @@ -12,26 +13,12 @@ export interface DatabaseUpdateStatus { export const updateDatabase = async ( client: FetchClient, + { start }: Metadata, _logger: Logger, -): Promise => { +): Promise => { const logger = _logger?.child({ label: "update_database" }); - const status: DatabaseUpdateStatus = { - updateStorage: 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; - } - - const { start } = metadata; - logger?.info(`Starting update from Q&A ${start}`); const { questions } = await fetchQuestionsIterative({ client, logger, @@ -39,7 +26,7 @@ export const updateDatabase = async ( }); if (questions.length === 0) { logger.info("No new questions to insert, returning."); - return status; + return []; } const [updateError, updates] = await updateQuestions(questions); @@ -48,10 +35,9 @@ export const updateDatabase = async ( { error: updateError }, `Failed to upsert ${questions.length} questions, retrying on next run.`, ); - return status; + return []; } - status.updateStorage = updates.length !== 0; logger.info(`${updates.length} new updates detected.`); - return status; + return updates; }; diff --git a/services/updater/src/update_forum_status.ts b/services/updater/src/update_forum_status.ts index 357a005..0853599 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, updateForumStates } from "@qnaplus/store"; +import { getAllPrograms, getForumStates, Metadata, updateForumStates } from "@qnaplus/store"; import type { Logger } from "pino"; type ProgramState = { @@ -23,6 +23,7 @@ const getNextSeason = (season: Season) => { export const updateForumStatus = async ( client: FetchClient, + { currentSeason: season }: Metadata, logger_: Logger, ) => { const logger = logger_.child({ label: "update_forum_status" }); @@ -35,14 +36,6 @@ export const updateForumStatus = async ( ); 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( @@ -56,7 +49,6 @@ export const updateForumStatus = async ( return map; }, {}); - const season = metadata.currentSeason; const newStates: ProgramState[] = []; for (const { program } of programs) { // get the 'open' state for a given program @@ -86,16 +78,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."); }; diff --git a/services/updater/src/update_storage.ts b/services/updater/src/update_storage.ts index 7bef9aa..5360848 100644 --- a/services/updater/src/update_storage.ts +++ b/services/updater/src/update_storage.ts @@ -1,56 +1,61 @@ import { getenv } from "@qnaplus/dotenv"; import { Question } from "@qnaplus/scraper"; -import { getAllQuestions, getAllSeasonQuestions, getMetadata, 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; } }; -export const updateStorage = async (_logger: Logger) => { - const logger = _logger?.child({ label: "update_storage" }); +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; +}; + +export const updateStorage = async ( + updates: Question[], + { 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`); + const ALL_QUESTIONS_KEY = `questions-${getenv("NODE_ENV")}.json`; + const SEASON_QUESTIONS_KEY = `questions-${getenv("NODE_ENV")}-${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.", - ); + const seasonQuestions = await get(SEASON_QUESTIONS_KEY, logger); + if (seasonQuestions === null) { return; } - const [metaError, meta] = await getMetadata(); - if (metaError || meta === undefined) { - logger?.error( - { error: metaError }, - "An error occurred while retreiving metadata from database.", - ); + 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; } - await update( - logger, - seasonQuestions, - `questions-${getenv("NODE_ENV")}-${meta.currentSeason}.json`, + 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."); };