feat(new-webui): Port search services from old-webui to Fastify plugins.

components/log-viewer-webui/server/settings.json

@@ -3,10 +3,12 @@
     "SqlDbPort": 3306,
     "SqlDbName": "clp-db",
     "SqlDbQueryJobsTableName": "query_jobs",
+
     "MongoDbHost": "localhost",
     "MongoDbPort": 27017,
     "MongoDbName": "clp-query-results",
     "MongoDbStreamFilesCollectionName": "stream-files",
+    "MongoDbSearchResultsMetadataCollectionName": "results-metadata",
 
     "ClientDir": "../../client/dist",
     "LogViewerDir": "../../yscope-log-viewer/dist",

components/log-viewer-webui/server/src/fastify-v2/app.ts

@@ -14,6 +14,7 @@ import FastifyV1App from "../app.js";
 const INTERNAL_SERVER_ERROR_CODE = 500;
 const RATE_LIMIT_MAX_REQUESTS = 3;
 const RATE_LIMIT_TIME_WINDOW_MS = 500;
+const IGNORED_FILES_REGEX = /^.*(?:utils|typings)\.js$/;
 
 /**
  * Registers all plugins and routes.
@@ -47,14 +48,16 @@ export default async function serviceApp (
     // Loads all application plugins.
     fastify.register(fastifyAutoload, {
         dir: path.join(import.meta.dirname, "plugins/app"),
+        ignorePattern: IGNORED_FILES_REGEX,
         options: {...opts},
     });
 
     // Loads all routes.
     fastify.register(fastifyAutoload, {
-        dir: path.join(import.meta.dirname, "routes"),
         autoHooks: true,
         cascadeHooks: true,
+        dir: path.join(import.meta.dirname, "routes"),
+        ignorePattern: IGNORED_FILES_REGEX,
         options: {...opts},
     });
 

components/log-viewer-webui/server/src/fastify-v2/plugins/app/search/QueryJobsDbManager/index.ts

@@ -0,0 +1,172 @@
+import type {MySQLPromisePool} from "@fastify/mysql";
+import {encode} from "@msgpack/msgpack";
+import settings from "@settings" with { type: "json" };
+import {FastifyInstance} from "fastify";
+import fp from "fastify-plugin";
+import {ResultSetHeader} from "mysql2";
+import {setTimeout} from "timers/promises";
+
+import {
+    QUERY_JOB_STATUS,
+    QUERY_JOB_STATUS_WAITING_STATES,
+    QUERY_JOB_TYPE,
+    QUERY_JOBS_TABLE_COLUMN_NAMES,
+    QueryJob,
+} from "../../../../../typings/query.js";
+import {JOB_COMPLETION_STATUS_POLL_INTERVAL_MILLIS} from "./typings.js";
+
+
+/**
+ * Class for submitting and monitoring query jobs in the database.
+ */
+class QueryJobsDbManager {
+    #sqlDbConnPool: MySQLPromisePool;
+
+    private constructor (sqlDbConnPool: MySQLPromisePool) {
+        this.#sqlDbConnPool = sqlDbConnPool;
+    }
+
+    /**
+     * Creates a new QueryJobsDbManager.
+     *
+     * @param fastify
+     * @return
+     */
+    static create (fastify: FastifyInstance): QueryJobsDbManager {
+        const sqlDbConnPool = fastify.mysql;
+        return new QueryJobsDbManager(sqlDbConnPool);
+    }
+
+    /**
+     * Submits a search job to the database.
+     *
+     * @param searchConfig The arguments for the query.
+     * @return The job's ID.
+     * @throws {Error} on error.
+     */
+    async submitSearchJob (searchConfig: object): Promise<number> {
+        const [queryInsertResults] = await this.#sqlDbConnPool.query<ResultSetHeader>(
+            `INSERT INTO ${settings.SqlDbQueryJobsTableName}
+           (${QUERY_JOBS_TABLE_COLUMN_NAMES.JOB_CONFIG},
+            ${QUERY_JOBS_TABLE_COLUMN_NAMES.TYPE})
+       VALUES (?, ?)`,
+            [
+                Buffer.from(encode(searchConfig)),
+                QUERY_JOB_TYPE.SEARCH_OR_AGGREGATION,
+            ]
+        );
+
+        return queryInsertResults.insertId;
+    }
+
+    /**
+     * Submits an aggregation job to the database.
+     *
+     * @param searchConfig The arguments for the query.
+     * @param timeRangeBucketSizeMillis
+     * @return The aggregation job's ID.
+     * @throws {Error} on error.
+     */
+    async submitAggregationJob (
+        searchConfig: object,
+        timeRangeBucketSizeMillis: number
+    ): Promise<number> {
+        const searchAggregationConfig = {
+            ...searchConfig,
+            aggregation_config: {
+                count_by_time_bucket_size: timeRangeBucketSizeMillis,
+            },
+        };
+
+        return await this.submitSearchJob(searchAggregationConfig);
+    }
+
+    /**
+     * Submits a query cancellation request to the database.
+     *
+     * @param jobId ID of the job to cancel.
+     * @return
+     * @throws {Error} on error.
+     */
+    async submitQueryCancellation (jobId: number): Promise<void> {
+        await this.#sqlDbConnPool.query(
+            `UPDATE ${settings.SqlDbQueryJobsTableName}
+                SET ${QUERY_JOBS_TABLE_COLUMN_NAMES.STATUS} = ${QUERY_JOB_STATUS.CANCELLING}
+                WHERE ${QUERY_JOBS_TABLE_COLUMN_NAMES.ID} = ?
+                AND ${QUERY_JOBS_TABLE_COLUMN_NAMES.STATUS}
+                IN (${QUERY_JOB_STATUS.PENDING}, ${QUERY_JOB_STATUS.RUNNING})`,
+            jobId,
+        );
+    }
+
+    /**
+     * Waits for the job to complete.
+     *
+     * @param jobId
+     * @return
+     * @throws {Error} on MySQL error, if the job wasn't found in the database, if the job was
+     * cancelled, or if the job completed in an unexpected state.
+     */
+    async awaitJobCompletion (jobId: number): Promise<void> {
+        // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+        while (true) {
+            let rows: QueryJob[];
+            try {
+                const [queryRows] = await this.#sqlDbConnPool.query<QueryJob[]>(
+                    `
+          SELECT ${QUERY_JOBS_TABLE_COLUMN_NAMES.STATUS}
+          FROM ${settings.SqlDbQueryJobsTableName}
+          WHERE ${QUERY_JOBS_TABLE_COLUMN_NAMES.ID} = ?
+          `,
+                    jobId
+                );
+
+                rows = queryRows;
+            } catch (e: unknown) {
+                let errorMessage: string;
+
+                if (e instanceof Error) {
+                    errorMessage = e.message;
+                } else {
+                    errorMessage = String(e);
+                }
+
+                throw new Error(`Failed to query status for job ${jobId} - ${errorMessage}`);
+            }
+            if (0 === rows.length) {
+                throw new Error(`Job ${jobId} not found in database.`);
+            }
+
+            const status = (rows[0] as QueryJob)[QUERY_JOBS_TABLE_COLUMN_NAMES.STATUS];
+
+            if (false === QUERY_JOB_STATUS_WAITING_STATES.has(status)) {
+                if (QUERY_JOB_STATUS.CANCELLED === status) {
+                    throw new Error(`Job ${jobId} was cancelled.`);
+                } else if (QUERY_JOB_STATUS.SUCCEEDED !== status) {
+                    throw new Error(
+                        `Job ${jobId} exited with unexpected status=${status}: ` +
+                        `${Object.keys(QUERY_JOB_STATUS)[status]}.`
+                    );
+                }
+                break;
+            }
+
+            await setTimeout(JOB_COMPLETION_STATUS_POLL_INTERVAL_MILLIS);
+        }
+    }
+}
+
+declare module "fastify" {
+    export interface FastifyInstance {
+        QueryJobsDbManager: QueryJobsDbManager;
+    }
+}
+
+export default fp(
+    (fastify) => {
+        fastify.decorate("QueryJobsDbManager", QueryJobsDbManager.create(fastify));
+    },
+    {
+        name: "QueryJobsDbManager",
+    }
+);

components/log-viewer-webui/server/src/fastify-v2/plugins/app/search/QueryJobsDbManager/typings.ts

@@ -0,0 +1,4 @@
+/**
+ * Interval in milliseconds for polling the completion status of a job.
+ */
+export const JOB_COMPLETION_STATUS_POLL_INTERVAL_MILLIS = 500;

components/log-viewer-webui/server/src/fastify-v2/plugins/app/search/SearchJobCollectionsManager/index.ts

@@ -0,0 +1,101 @@
+import {FastifyInstance} from "fastify";
+import fp from "fastify-plugin";
+import type {
+    Collection,
+    Db,
+} from "mongodb";
+
+import {Nullable} from "../../../../../typings/common.js";
+import {
+    CollectionDroppedError,
+    SearchResultsDocument,
+} from "./typings.js";
+
+
+/**
+ * Class to keep track of MongoDB collections created for search jobs, ensuring all collections
+ * have unique names.
+ */
+class SearchJobCollectionsManager {
+    #jobIdToCollectionsMap: Map<string, Nullable<Collection<SearchResultsDocument>>> =
+        new Map();
+
+    #db: Db;
+
+    private constructor (db: Db) {
+        this.#db = db;
+    }
+
+    /**
+     * Creates a new SearchJobCollectionsManager.
+     *
+     * @param fastify
+     * @return
+     * @throws {Error} if the MongoDB database is not found.
+     */
+    static create (fastify: FastifyInstance): SearchJobCollectionsManager {
+        if ("undefined" === typeof fastify.mongo.db) {
+            throw new Error("MongoDB database not found");
+        }
+
+        return new SearchJobCollectionsManager(fastify.mongo.db);
+    }
+
+    /**
+     * Gets, or if it doesn't exist, creates a MongoDB collection named with the given job ID.
+     *
+     * @param jobId
+     * @return MongoDB collection
+     * @throws {CollectionDroppedError} if the collection was already dropped.
+     */
+    async getOrCreateCollection (jobId: number): Promise<Collection<SearchResultsDocument>> {
+        const name = jobId.toString();
+        if (false === this.#jobIdToCollectionsMap.has(name)) {
+            this.#jobIdToCollectionsMap.set(name, this.#db.collection(name));
+        } else if (null === this.#jobIdToCollectionsMap.get(name)) {
+            throw new CollectionDroppedError(name);
+        }
+
+
+        // `collections.get(name)` will always return a valid collection since:
+        // -  Function creates a new collection if it doesn't exist in map.
+        // -  Throws an error if the collection was already dropped.
+        // Therefore, we can safely cast to `Collection<SearchResultsDocument>`.
+        return this.#jobIdToCollectionsMap.get(name) as Collection<SearchResultsDocument>;
+    }
+
+    /**
+     * Drops the MongoDB collection associated with the given job ID.
+     *
+     * @param jobId
+     * @throws {Error} if the collection does not exist or has already been dropped.
+     */
+    async dropCollection (jobId: number) {
+        const name = jobId.toString();
+        const collection = this.#jobIdToCollectionsMap.get(name);
+        if ("undefined" === typeof collection || null === collection) {
+            throw new Error(`Collection ${name} not found`);
+        }
+
+        await collection.drop();
+        this.#jobIdToCollectionsMap.set(name, null);
+    }
+}
+
+declare module "fastify" {
+    export interface FastifyInstance {
+        SearchJobCollectionsManager: SearchJobCollectionsManager;
+    }
+}
+
+export default fp(
+    (fastify) => {
+        fastify.decorate(
+            "SearchJobCollectionsManager",
+            SearchJobCollectionsManager.create(fastify)
+        );
+    },
+    {
+        name: "SearchJobCollectionsManager",
+    }
+);

components/log-viewer-webui/server/src/fastify-v2/plugins/app/search/SearchJobCollectionsManager/typings.ts

@@ -0,0 +1,25 @@
+/**
+ * MongoDB document for search results.
+ */
+interface SearchResultsDocument {
+    _id: string;
+    orig_file_id: string;
+    orig_file_path: string;
+    log_event_ix: number;
+    timestamp: number;
+    message: string;
+}
+
+/**
+ * Error thrown when a MongoDB collection has been dropped unexpectedly.
+ */
+class CollectionDroppedError extends Error {
+    constructor (collectionName: string) {
+        super(`Collection ${collectionName} has been dropped.`);
+        this.name = "CollectionDroppedError";
+    }
+}
+
+export {
+    CollectionDroppedError, SearchResultsDocument,
+};

components/log-viewer-webui/server/src/fastify-v2/plugins/app/search/SearchResultsMetadataCollection/index.ts

@@ -0,0 +1,41 @@
+import settings from "@settings" with {type: "json"};
+import {FastifyInstance} from "fastify";
+import fp from "fastify-plugin";
+
+import type {SearchResultsMetadataDocument} from "./typings.js";
+
+
+/**
+ * Creates a MongoDB collection for search results metadata.
+ *
+ * @param fastify
+ * @return MongoDB collection
+ * @throws {Error} if the MongoDB database is not found.
+ */
+const createSearchResultsMetadataCollection = (fastify: FastifyInstance) => {
+    if ("undefined" === typeof fastify.mongo.db) {
+        throw new Error("MongoDB database not found");
+    }
+
+    return fastify.mongo.db.collection<SearchResultsMetadataDocument>(
+        settings.MongoDbSearchResultsMetadataCollectionName
+    );
+};
+
+declare module "fastify" {
+    export interface FastifyInstance {
+        SearchResultsMetadataCollection: ReturnType<typeof createSearchResultsMetadataCollection>;
+    }
+}
+
+export default fp(
+    (fastify) => {
+        fastify.decorate(
+            "SearchResultsMetadataCollection",
+            createSearchResultsMetadataCollection(fastify)
+        );
+    },
+    {
+        name: "SearchResultsMetadataCollection",
+    }
+);