fix

Co-authored-by: Johann150 <johann@qwertqwefsday.eu>

.config/example.yml

@@ -71,6 +71,16 @@ redis:
 #  user: 
 #  pass: 
 
+
+#   ┌─────────────────────┐
+#───┘ Sonic configuration └─────────────────────────────
+
+#sonic:
+#  host: localhost
+#  port: 1491
+#  pass: example-pass
+
+
 #   ┌───────────────┐
 #───┘ ID generation └───────────────────────────────────────────
 

.gitignore

@@ -42,3 +42,5 @@ ormconfig.json
 *.blend3
 *.blend4
 *.blend5
+
+.yarn

packages/backend/package.json

@@ -14,6 +14,7 @@
 		"lodash": "^4.17.21"
 	},
 	"dependencies": {
+		"@bull-board/koa": "3.10.2",
 		"@discordapp/twemoji": "13.1.1",
 		"@elastic/elasticsearch": "7.11.0",
 		"@koa/cors": "3.1.0",
@@ -67,7 +68,6 @@
 		"@types/ws": "8.5.3",
 		"@typescript-eslint/eslint-plugin": "5.17.0",
 		"@typescript-eslint/parser": "5.17.0",
-		"@bull-board/koa": "3.10.2",
 		"abort-controller": "3.0.0",
 		"ajv": "8.11.0",
 		"archiver": "5.3.0",
@@ -145,8 +145,10 @@
 		"rndstr": "1.0.0",
 		"s-age": "1.1.2",
 		"sanitize-html": "2.7.0",
+		"seedrandom": "3.0.5",
 		"semver": "7.3.5",
 		"sharp": "0.30.3",
+		"sonic-channel": "1.2.6",
 		"speakeasy": "2.0.0",
 		"strict-event-emitter-types": "2.0.0",
 		"stringz": "2.1.0",

packages/backend/src/@types/sonic-channel.d.ts

@@ -0,0 +1,33 @@
+declare module 'sonic-channel' {
+	type ConnectionListeners = {
+		connected?: () => void,
+		disconnected?: (err: Error | null) => void,
+		timeout?: () => void,
+		retrying?: () => void,
+		error?: (err: Error) => void,
+	};
+
+	type ConnectionParams = {
+		host: string,
+		port: number,
+		auth: string | null
+	};
+
+	class Ingest {
+		constructor(options: ConnectionParams);
+
+		public connect(handlers: ConnectionListeners): Ingest;
+		public close(): Promise<void>;
+
+		public push(collection_id: string, bucket_id: string, object_id: string, text: string, lang?: string): Promise<void>;
+	}
+
+	class Search {
+		constructor(options: ConnectionParams);
+
+		public connect(handlers: ConnectionListeners): Search;
+		public close(): Promise<void>;
+
+		public query(collection_id: string, bucket_id: string, terms_text: string, limit?: number, offset?: number, lang?: string): Promise<string[]>;
+	}
+}

packages/backend/src/config/types.ts

@@ -23,7 +23,7 @@ export type Source = {
 		db?: number;
 		prefix?: string;
 	};
-	elasticsearch: {
+	elasticsearch?: {
 		host: string;
 		port: number;
 		ssl?: boolean;
@@ -31,6 +31,12 @@ export type Source = {
 		pass?: string;
 		index?: string;
 	};
+	sonic?: {
+		host: string;
+		port: number;
+		pass: string;
+		index?: string;
+	};
 
 	proxy?: string;
 	proxySmtp?: string;

packages/backend/src/db/SearchClientBase.ts

@@ -0,0 +1,3 @@
+import { EventEmitter } from 'events';
+
+export class SearchClientBase extends EventEmitter { }

packages/backend/src/db/elasticsearch.ts

@@ -1,7 +1,10 @@
 import * as elasticsearch from '@elastic/elasticsearch';
+import config from '../config';
+import { SearchClientBase } from './SearchClientBase';
+import { Note } from '../models/entities/note';
 import config from '@/config/index.js';
 
-const index = {
+const indexData = {
 	settings: {
 		analysis: {
 			analyzer: {
@@ -30,27 +33,107 @@ const index = {
 	},
 };
 
-// Init ElasticSearch connection
-const client = config.elasticsearch ? new elasticsearch.Client({
-	node: `${config.elasticsearch.ssl ? 'https://' : 'http://'}${config.elasticsearch.host}:${config.elasticsearch.port}`,
-	auth: (config.elasticsearch.user && config.elasticsearch.pass) ? {
-		username: config.elasticsearch.user,
-		password: config.elasticsearch.pass,
-	} : undefined,
-	pingTimeout: 30000,
-}) : null;
+class ElasticSearch extends SearchClientBase {
+	public index = 'misskey_note';
+	constructor(address: string, index?: string) {
+		super();
+		this.index = index || 'misskey_note';
+		// Init ElasticSearch connection
+		this._client = new elasticsearch.Client({
+			node: address,
+			pingTimeout: 30000,
+		});
 
-if (client) {
-	client.indices.exists({
-		index: config.elasticsearch.index || 'misskey_note',
-	}).then(exist => {
-		if (!exist.body) {
-			client.indices.create({
-				index: config.elasticsearch.index || 'misskey_note',
-				body: index,
+		this._client.indices
+			.exists({
+				index: this.index,
+			})
+			.then(exist => {
+				if (!exist.body) {
+					this._client.indices.create({
+						index: this.index,
+						body: indexData,
+					});
+				}
 			});
+	}
+
+	private _client: elasticsearch.Client;
+
+	public available = true;
+
+	public search(
+		content: string,
+		qualifiers: {userId?: string | null; userHost?: string | null} = {},
+		limit?: number,
+		offset?: number,
+	) {
+		const queries: any[] = [
+			{
+				simple_query_string: {
+					fields: ['text'],
+					query: content.toLowerCase(),
+					default_operator: 'and',
+				},
+			},
+		];
+
+		if (qualifiers.userId) {
+			queries.push({
+				term: { userId: qualifiers.userId },
+			});
+		} else if (qualifiers.userHost !== undefined) {
+			if (qualifiers.userHost === null) {
+				queries.push({
+					bool: {
+						must_not: {
+							exists: {
+								field: 'userHost',
+							},
+						},
+					},
+				});
+			} else {
+				queries.push({
+					term: {
+						userHost: qualifiers.userHost,
+					},
+				});
+			}
 		}
-	});
+
+		return this._client
+			.search({
+				index: this.index,
+				body: {
+					size: limit,
+					from: offset,
+					query: {
+						bool: {
+							must: queries,
+						},
+					},
+				},
+			})
+			.then(result => result.body.hits.hits.map((hit: any) => hit._id));
+	}
+
+	public push(note: Note) {
+		return this._client.index({
+			index: this.index,
+			id: note.id.toString(),
+			body: {
+				text: String(note.text).toLowerCase(),
+				userId: note.userId,
+				userHost: note.userHost,
+			},
+		});
+	}
 }
 
-export default client;
+export default (config.elasticsearch
+	? new ElasticSearch(
+		`${config.elasticsearch.ssl ? 'https://' : 'http://'}${config.elasticsearch.host}:${config.elasticsearch.port}`,
+		config.elasticsearch.index,
+	)
+	: null);

packages/backend/src/db/searchClient.ts

@@ -0,0 +1,11 @@
+import sonic from "./sonic";
+import es from "./elasticsearch";
+
+// This file is just to make it easier to add new drivers in the future, simply import searchClient and whatever driver is available is used
+
+export const clients = [sonic, es];
+
+const searchClient =
+	clients.find((client) => client && client.available) || null;
+
+export default searchClient;

packages/backend/src/db/sonic.ts

@@ -0,0 +1,158 @@
+import * as Sonic from 'sonic-channel';
+import config from '../config';
+import { SearchClientBase } from './SearchClientBase';
+import { Note } from '../models/entities/note';
+
+export class SonicDriver extends SearchClientBase {
+	public available = true;
+	public index = 'misskey_note';
+	public locale = 'none';
+
+	public _ingestQueue: (() => Promise<void>)[] = [];
+	public _searchQueue: (() => Promise<void>)[] = [];
+	public _searchReady = false;
+	public _ingestReady = false;
+
+	public _ingestClient: Sonic.Ingest;
+	public _searchClient: Sonic.Search;
+
+	constructor(connectionArgs: {
+		host: string;
+		port: number;
+		auth: string | null;
+	}, index?: string) {
+		super();
+		// Bad!
+		const self = this;
+		this.index = index || 'misskey_note';
+		this._ingestClient = new Sonic.Ingest(connectionArgs).connect({
+			connected() {
+				// execute queue of queries
+				self._runIngestQueue();
+				self._ingestReady = true;
+				self._emitReady();
+			},
+			disconnected() {
+				self._ingestReady = false;
+				self.emit('disconnected');
+			},
+			timeout() { },
+			retrying() { },
+			error(err: Error) {
+				self.emit('error', err);
+			},
+		});
+
+		self._searchClient = new Sonic.Search(connectionArgs).connect({
+			connected() {
+				// execute queue of queries
+				self._runSearchQueue();
+				self._searchReady = true;
+				self._emitReady();
+			},
+			disconnected() {
+				self._searchReady = false;
+				self.emit('disconnected');
+			},
+			timeout() { },
+			retrying() { },
+			error(err: Error) {
+				self.emit('error', err);
+			},
+		});
+	}
+
+	get ready() {
+		return this._searchReady && this._ingestReady;
+	}
+	public _emitReady() {
+		if (this.ready) this.emit('ready');
+	}
+	public async disconnect() {
+		return await Promise.all([
+			this._searchClient.close(),
+			this._ingestClient.close(),
+		]);
+	}
+
+	public search(
+		content: string,
+		qualifiers: { userId?: string | null; userHost?: string | null } = {},
+		limit: number = 20,
+		offset?: number,
+		locale?: string,
+	) {
+		const doSearch = () =>
+			this._searchClient.query(
+				this.index,
+				pickQualifier(qualifiers),
+				content,
+				limit,
+				offset,
+				locale || this.locale,
+			);
+
+		if (this._searchReady) {
+			return doSearch();
+		} else {
+			return new Promise((resolve, reject) => {
+				this._searchQueue.push(() =>
+					doSearch()
+						.then(resolve)
+						.catch(reject),
+				);
+			});
+		}
+	}
+
+	public push(note: Note) {
+		const doIngest = () => {
+			return Promise.all(
+				['userId-' + note.userId, 'userHost-' + note.userHost, 'default']
+					.map((bucket: string) =>
+						this._ingestClient.push(
+							this.index,
+							bucket,
+							note.id,
+							String(note.text).toLowerCase(),
+							this.locale,
+						),
+					),
+			);
+		};
+
+		if (this._ingestReady) {
+			return doIngest();
+		} else {
+			return new Promise((resolve, reject) => {
+				this._ingestQueue.push(() =>
+					doIngest()
+						.then(resolve)
+						.catch(reject),
+				);
+			});
+		}
+	}
+
+	public _runIngestQueue() {
+		return Promise.all(this._ingestQueue.map(cb => cb()));
+	}
+
+	public _runSearchQueue() {
+		return Promise.all(this._searchQueue.map(cb => cb()));
+	}
+}
+
+function pickQualifier(qualifiers: { userId?: string | null; userHost?: string | null }) {
+	if (qualifiers.userId) return 'userId-' + qualifiers.userId;
+	else if (qualifiers.userHost) return 'userHost-' + qualifiers.userHost;
+	else return 'default';
+}
+
+export default (config.sonic
+	? new SonicDriver({
+		host: config.sonic.host,
+		port: config.sonic.port,
+		auth: config.sonic.pass == undefined ? null : config.sonic.pass
+	}, config.sonic.index)
+	: null);

packages/backend/src/server/api/endpoints/notes/search.ts

@@ -1,75 +1,86 @@
-import es from '../../../../db/elasticsearch.js';
-import define from '../../define.js';
-import { Notes } from '@/models/index.js';
-import { In } from 'typeorm';
-import config from '@/config/index.js';
-import { makePaginationQuery } from '../../common/make-pagination-query.js';
-import { generateVisibilityQuery } from '../../common/generate-visibility-query.js';
-import { generateMutedUserQuery } from '../../common/generate-muted-user-query.js';
-import { generateBlockedUserQuery } from '../../common/generate-block-query.js';
+import searchClient from "@/db/searchClient";
+import define from "../../define";
+import { Notes } from "@/models/index";
+import { In } from "typeorm";
+import config from "@/config/index.js";
+import { makePaginationQuery } from "../../common/make-pagination-query.js";
+import { generateVisibilityQuery } from "../../common/generate-visibility-query.js";
+import { generateMutedUserQuery } from "../../common/generate-muted-user-query.js";
+import { generateBlockedUserQuery } from "../../common/generate-block-query.js";
 
 export const meta = {
-	tags: ['notes'],
+	tags: ["notes"],
 
 	requireCredential: false,
 
 	res: {
-		type: 'array',
-		optional: false, nullable: false,
+		type: "array",
+		optional: false,
+		nullable: false,
 		items: {
-			type: 'object',
-			optional: false, nullable: false,
-			ref: 'Note',
+			type: "object",
+			optional: false,
+			nullable: false,
+			ref: "Note",
 		},
 	},
 
-	errors: {
-	},
+	errors: {},
 } as const;
 
 export const paramDef = {
-	type: 'object',
+	type: "object",
 	properties: {
-		query: { type: 'string' },
-		sinceId: { type: 'string', format: 'misskey:id' },
-		untilId: { type: 'string', format: 'misskey:id' },
-		limit: { type: 'integer', minimum: 1, maximum: 100, default: 10 },
-		offset: { type: 'integer', default: 0 },
+		query: { type: "string" },
+		sinceId: { type: "string", format: "misskey:id" },
+		untilId: { type: "string", format: "misskey:id" },
+		limit: { type: "integer", minimum: 1, maximum: 100, default: 10 },
+		offset: { type: "integer", default: 0 },
 		host: {
-			type: 'string',
+			type: "string",
 			nullable: true,
-			description: 'The local host is represented with `null`.',
+			description: "The local host is represented with `null`.",
+		},
+		userId: {
+			type: "string",
+			format: "misskey:id",
+			nullable: true,
+			default: null,
+		},
+		channelId: {
+			type: "string",
+			format: "misskey:id",
+			nullable: true,
+			default: null,
 		},
-		userId: { type: 'string', format: 'misskey:id', nullable: true, default: null },
-		channelId: { type: 'string', format: 'misskey:id', nullable: true, default: null },
 	},
-	required: ['query'],
+	required: ["query"],
 } as const;
 
 // eslint-disable-next-line import/no-default-export
-export default define(meta, paramDef, async (ps, me) => {
-	if (es == null) {
-		const query = makePaginationQuery(Notes.createQueryBuilder('note'), ps.sinceId, ps.untilId);
+export default define(meta, async (ps, me, cb) => {
+	if (searchClient == null) {
+		const query = makePaginationQuery(
+			Notes.createQueryBuilder("note"),
+			ps.sinceId,
+			ps.untilId
+		);
 
 		if (ps.userId) {
-			query.andWhere('note.userId = :userId', { userId: ps.userId });
+			query.andWhere("note.userId = :userId", { userId: ps.userId });
 		} else if (ps.channelId) {
-			query.andWhere('note.channelId = :channelId', { channelId: ps.channelId });
+			query.andWhere("note.channelId = :channelId", {
+				channelId: ps.channelId,
+			});
 		}
 
 		query
-			.andWhere('note.text ILIKE :q', { q: `%${ps.query}%` })
-			.innerJoinAndSelect('note.user', 'user')
-			.leftJoinAndSelect('user.avatar', 'avatar')
-			.leftJoinAndSelect('user.banner', 'banner')
-			.leftJoinAndSelect('note.reply', 'reply')
-			.leftJoinAndSelect('note.renote', 'renote')
-			.leftJoinAndSelect('reply.user', 'replyUser')
-			.leftJoinAndSelect('replyUser.avatar', 'replyUserAvatar')
-			.leftJoinAndSelect('replyUser.banner', 'replyUserBanner')
-			.leftJoinAndSelect('renote.user', 'renoteUser')
-			.leftJoinAndSelect('renoteUser.avatar', 'renoteUserAvatar')
-			.leftJoinAndSelect('renoteUser.banner', 'renoteUserBanner');
+			.andWhere("note.text ILIKE :q", { q: `%${ps.query}%` })
+			.innerJoinAndSelect("note.user", "user")
+			.leftJoinAndSelect("note.reply", "reply")
+			.leftJoinAndSelect("note.renote", "renote")
+			.leftJoinAndSelect("reply.user", "replyUser")
+			.leftJoinAndSelect("renote.user", "renoteUser");
 
 		generateVisibilityQuery(query, me);
 		if (me) generateMutedUserQuery(query, me);
@@ -79,55 +90,13 @@ export default define(meta, paramDef, async (ps, me) => {
 
 		return await Notes.packMany(notes, me);
 	} else {
-		const userQuery = ps.userId != null ? [{
-			term: {
-				userId: ps.userId,
-			},
-		}] : [];
-
-		const hostQuery = ps.userId == null ?
-			ps.host === null ? [{
-				bool: {
-					must_not: {
-						exists: {
-							field: 'userHost',
-						},
-					},
-				},
-			}] : ps.host !== undefined ? [{
-				term: {
-					userHost: ps.host,
-				},
-			}] : []
-		: [];
-
-		const result = await es.search({
-			index: config.elasticsearch.index || 'misskey_note',
-			body: {
-				size: ps.limit,
-				from: ps.offset,
-				query: {
-					bool: {
-						must: [{
-							simple_query_string: {
-								fields: ['text'],
-								query: ps.query.toLowerCase(),
-								default_operator: 'and',
-							},
-						}, ...hostQuery, ...userQuery],
-					},
-				},
-				sort: [{
-					_doc: 'desc',
-				}],
-			},
+		const hits = await searchClient.search(ps.query, {
+			userHost: ps.host,
+			userId: ps.userId,
 		});
 
-		const hits = result.body.hits.hits.map((hit: any) => hit._id);
-
 		if (hits.length === 0) return [];
 
-		// Fetch found notes
 		const notes = await Notes.find({
 			where: {
 				id: In(hits),

packages/backend/src/services/note/create.ts

@@ -1,4 +1,24 @@
 import * as mfm from 'mfm-js';
+import searchClient from '@/db/searchClient';
+import { publishMainStream, publishNotesStream } from '@/services/stream';
+import DeliverManager from '@/remote/activitypub/deliver-manager';
+import renderNote from '@/remote/activitypub/renderer/note';
+import renderCreate from '@/remote/activitypub/renderer/create';
+import renderAnnounce from '@/remote/activitypub/renderer/announce';
+import { renderActivity } from '@/remote/activitypub/renderer/index';
+import { resolveUser } from '@/remote/resolve-user';
+import config from '@/config/index';
+import { updateHashtags } from '../update-hashtag';
+import { concat } from '@/prelude/array';
+import { insertNoteUnread } from '@/services/note/unread';
+import { registerOrFetchInstanceDoc } from '../register-or-fetch-instance-doc';
+import { extractMentions } from '@/misc/extract-mentions';
+import { extractCustomEmojisFromMfm } from '@/misc/extract-custom-emojis-from-mfm';
+import { extractHashtags } from '@/misc/extract-hashtags';
+import { Note, IMentionedRemoteUsers } from '@/models/entities/note';
+import { Mutings, Users, NoteWatchings, Notes, Instances, UserProfiles, Antennas, Followings, MutedNotes, Channels, ChannelFollowings, Blockings, NoteThreadMutings } from '@/models/index';
+import { DriveFile } from '@/models/entities/drive-file';
+import { App } from '@/models/entities/app';
 import es from '../../db/elasticsearch.js';
 import { publishMainStream, publishNotesStream } from '@/services/stream.js';
 import DeliverManager from '@/remote/activitypub/deliver-manager.js';
@@ -593,17 +613,9 @@ async function insertNote(user: { id: User['id']; host: User['host']; }, data: O
 }
 
 function index(note: Note) {
-	if (note.text == null || config.elasticsearch == null) return;
+	if (note.text == null || searchClient == null) return;
 
-	es!.index({
-		index: config.elasticsearch.index || 'misskey_note',
-		id: note.id.toString(),
-		body: {
-			text: normalizeForSearch(note.text),
-			userId: note.userId,
-			userHost: note.userHost,
-		},
-	});
+	return searchClient.push(note);
 }
 
 async function notifyToWatchersOfRenotee(renote: Note, user: { id: User['id']; }, nm: NotificationManager, type: NotificationType) {