feat: search with sonic

packages/sonic/model.ts

@@ -0,0 +1,27 @@
+import * as bus from 'hydrooj/dist/service/bus';
+import * as system from 'hydrooj/dist/model/system';
+import sonic from './service';
+
+bus.on('problem/add', async (doc, docId) => {
+    await sonic.push('problem', `${doc.domainId}@title`, docId.toString(), doc.title);
+    await sonic.push('problem', `${doc.domainId}@content`, docId.toString(), doc.content.toString());
+});
+
+bus.on('problem/edit', async (pdoc) => {
+    await sonic.flusho('problem', `${pdoc.domainId}@title`, pdoc.docId.toString());
+    await sonic.flusho('problem', `${pdoc.domainId}@content`, pdoc.docId.toString());
+    await sonic.push('problem', `${pdoc.domainId}@title`, pdoc.docId.toString(), pdoc.title);
+    await sonic.push('problem', `${pdoc.domainId}@content`, pdoc.docId.toString(), pdoc.content.toString());
+});
+
+bus.on('problem/del', async (domainId, docId) => {
+    await sonic.flusho('problem', `${domainId}@title`, docId.toString());
+    await sonic.flusho('problem', `${domainId}@content`, docId.toString());
+});
+
+global.Hydro.lib.problemSearch = async (domainId: string, query: string) => {
+    const c = system.get('pagination.problem');
+    const docIds = await sonic.query('problem', `${domainId}@title`, query, { limit: c });
+    docIds.push(...await sonic.query('problem', `${domainId}@content`, query, { limit: c - docIds.length }));
+    return docIds.map(Number);
+};

packages/sonic/package.json

@@ -0,0 +1,13 @@
+{
+    "name": "@hydrooj/sonic",
+    "version": "1.0.0",
+    "description": "Sonic search service",
+    "main": "service.js",
+    "typings": "service.d.ts",
+    "repository": "https://github.com/hydro-dev/Hydro.git",
+    "author": "undefined <i@undefined.moe>",
+    "license": "AGPL-3.0-only",
+    "dependencies": {
+        "sonic-channel": "^1.2.6"
+    }
+}

packages/sonic/script.ts

@@ -0,0 +1,25 @@
+import { iterateAllProblem, iterateAllProblemInDomain } from 'hydrooj/dist/upgrade';
+import sonic from './service';
+
+export const description = 'Sonic problem search re-index';
+
+export async function run({ domainId }) {
+    if (domainId) await sonic.flushb('problem', domainId);
+    else await sonic.flushc('problem');
+    const cb = async (pdoc) => {
+        await sonic.push('problem', `${pdoc.domainId}@title`, pdoc.docId.toString(), pdoc.title);
+        await sonic.push('problem', `${pdoc.domainId}@content`, pdoc.docId.toString(), pdoc.content.toString());
+    };
+    if (domainId) await iterateAllProblemInDomain(domainId, ['title', 'content'], cb);
+    else await iterateAllProblem(['title', 'content'], cb);
+    return true;
+}
+
+export const validate = {
+    $or: [
+        { domainId: 'string' },
+        { domainId: 'undefined' },
+    ],
+};
+
+global.Hydro.script.ensureSonicSearch = { run, description, validate };

packages/sonic/service.ts

@@ -0,0 +1,117 @@
+import { BaseService } from 'hydrooj';
+import * as system from 'hydrooj/dist/model/system';
+import {
+    Search, Ingest, QueryOptions, SuggestOptions, PushOptions,
+} from 'sonic-channel';
+import { Logger } from 'hydrooj/dist/logger';
+
+const logger = new Logger('sonic');
+
+declare module 'hydrooj/dist/interface' {
+    interface SystemKeys {
+        'sonic.host': string;
+        'sonic.port': number;
+        'sonic.auth': string;
+    }
+}
+
+function getHandler(type: string, that: any) {
+    return {
+        connected() {
+            that.started = true;
+            logger.info(`Sonic Channel succeeded to connect to host (${type}).`);
+        },
+        disconnected() {
+            that.started = false;
+            logger.error(`Sonic Channel is now disconnected (${type}).`);
+        },
+        timeout() {
+            logger.error(`Sonic Channel connection timed out (${type}).`);
+        },
+        retrying() {
+            logger.error(`Trying to reconnect to Sonic Channel (${type})...`);
+        },
+        error(error) {
+            that.error = error;
+            logger.error(`Sonic Channel failed to connect to host (${type}).`, error);
+        },
+    };
+}
+
+class SonicService implements BaseService {
+    public search: Search;
+    public ingest: Ingest;
+    public started = false;
+    public error = '';
+
+    async start() {
+        const [host, port, auth] = system.getMany(['sonic.host', 'sonic.port', 'sonic.auth']);
+        const cfg = {
+            host: host || '127.0.0.1',
+            port: port || 1491,
+            auth: auth || 'sonic-passwd',
+        };
+        this.search = new Search(cfg);
+        this.ingest = new Ingest(cfg);
+        await this.connect();
+    }
+
+    async connect() {
+        try {
+            this.search.connect(getHandler('search', this));
+            this.ingest.connect(getHandler('ingest', this));
+        } catch (e) {
+            logger.warn('Sonic init fail. will retry later.');
+            this.error = e.toString();
+            setTimeout(() => this.start(), 10000);
+        }
+    }
+
+    async query(collection: string, bucket: string, terms: string, options?: QueryOptions) {
+        return await this.search.query(collection, bucket, terms, options);
+    }
+
+    async suggest(collection: string, bucket: string, word: string, options?: SuggestOptions) {
+        return await this.search.suggest(collection, bucket, word, options);
+    }
+
+    async push(collection: string, bucket: string, object: string, text: string, options?: PushOptions) {
+        return await this.ingest.push(collection, bucket, object, text, options);
+    }
+
+    async pop(collection: string, bucket: string, object: string, text: string) {
+        return await this.ingest.pop(collection, bucket, object, text);
+    }
+
+    async count(collection: string, bucket?: string, object?: string) {
+        return await this.ingest.count(collection, bucket, object);
+    }
+
+    async flusho(collection: string, bucket: string, object: string) {
+        return await this.ingest.flusho(collection, bucket, object);
+    }
+
+    async flushb(collection: string, bucket: string) {
+        return await this.ingest.flushb(collection, bucket);
+    }
+
+    async flushc(collection: string) {
+        return await this.ingest.flushc(collection);
+    }
+
+    async stop() {
+        await Promise.all([
+            this.search.close(),
+            this.ingest.close(),
+        ]);
+    }
+}
+
+const service = new SonicService();
+global.Hydro.service.sonic = service;
+export = service;
+declare module 'hydrooj/dist/interface' {
+    interface Service {
+        sonic: typeof service
+    }
+}

packages/sonic/setting.yaml

@@ -0,0 +1,7 @@
+host:
+  value: localhost
+port:
+  type: number
+  value: 1491
+auth:
+  type: text