Make it a daemon, add reply function

This commit is contained in:
2024-12-28 23:47:33 +00:00
parent df1803d864
commit bb08f5f800
8 changed files with 948 additions and 70 deletions

470
misskey-js/streaming.ts Normal file
View File

@@ -0,0 +1,470 @@
import { EventEmitter } from "eventemitter3";
import ReconnectingWebsocket, {
type Options as WebsocketOptions,
} from "reconnecting-websocket";
import type { BroadcastEvents, Channels } from "./streaming.types.js";
export function urlQuery(
obj: Record<string, string | number | boolean | undefined>,
): string {
const params = Object.entries(obj)
.filter(([, v]) => (Array.isArray(v) ? v.length : v !== undefined))
.reduce(
// biome-ignore lint/style/noNonNullAssertion: <explanation>
// biome-ignore lint/suspicious/noAssignInExpressions: <explanation>
// biome-ignore lint/style/noCommaOperator: <explanation>
(a, [k, v]) => ((a[k] = v!), a),
{} as Record<string, string | number | boolean>,
);
return Object.entries(params)
.map((e) => `${e[0]}=${encodeURIComponent(e[1])}`)
.join("&");
}
type AnyOf<T extends Record<PropertyKey, unknown>> = T[keyof T];
export type StreamEvents = {
_connected_: undefined;
_disconnected_: undefined;
} & BroadcastEvents;
export interface IStream extends EventEmitter<StreamEvents> {
state: "initializing" | "reconnecting" | "connected";
useChannel<C extends keyof Channels>(
channel: C,
params?: Channels[C]["params"],
name?: string,
): IChannelConnection<Channels[C]>;
removeSharedConnection(connection: SharedConnection): void;
removeSharedConnectionPool(pool: Pool): void;
disconnectToChannel(connection: NonSharedConnection): void;
send(typeOrPayload: string): void;
send(typeOrPayload: string, payload: unknown): void;
send(typeOrPayload: Record<string, unknown> | unknown[]): void;
send(
typeOrPayload: string | Record<string, unknown> | unknown[],
payload?: unknown,
): void;
ping(): void;
heartbeat(): void;
close(): void;
}
/**
* Misskey stream connection
*/
// eslint-disable-next-line import/no-default-export
export default class Stream
extends EventEmitter<StreamEvents>
implements IStream {
private stream: ReconnectingWebsocket;
public state: "initializing" | "reconnecting" | "connected" = "initializing";
private sharedConnectionPools: Pool[] = [];
private sharedConnections: SharedConnection[] = [];
private nonSharedConnections: NonSharedConnection[] = [];
private idCounter = 0;
constructor(
origin: string,
user: { token: string } | null,
options?: WebsocketOptions,
) {
super();
this.genId = this.genId.bind(this);
this.useChannel = this.useChannel.bind(this);
this.useSharedConnection = this.useSharedConnection.bind(this);
this.removeSharedConnection = this.removeSharedConnection.bind(this);
this.removeSharedConnectionPool =
this.removeSharedConnectionPool.bind(this);
this.connectToChannel = this.connectToChannel.bind(this);
this.disconnectToChannel = this.disconnectToChannel.bind(this);
this.onOpen = this.onOpen.bind(this);
this.onClose = this.onClose.bind(this);
this.onMessage = this.onMessage.bind(this);
this.send = this.send.bind(this);
this.close = this.close.bind(this);
const query = urlQuery({
i: user?.token,
// To prevent cache of an HTML such as error screen
_t: Date.now(),
});
const wsOrigin = origin
.replace("http://", "ws://")
.replace("https://", "wss://");
this.stream = new ReconnectingWebsocket(
`${wsOrigin}/streaming?${query}`,
"",
{
minReconnectionDelay: 1, // https://github.com/pladaria/reconnecting-websocket/issues/91
...(options ?? {}),
},
);
this.stream.binaryType = "arraybuffer";
this.stream.addEventListener("open", this.onOpen);
this.stream.addEventListener("close", this.onClose);
this.stream.addEventListener("message", this.onMessage);
}
private genId(): string {
return (++this.idCounter).toString();
}
public useChannel<C extends keyof Channels>(
channel: C,
params?: Channels[C]["params"],
name?: string,
): Connection<Channels[C]> {
if (params) {
return this.connectToChannel(channel, params);
}
return this.useSharedConnection(channel, name);
}
private useSharedConnection<C extends keyof Channels>(
channel: C,
name?: string,
): SharedConnection<Channels[C]> {
let pool = this.sharedConnectionPools.find((p) => p.channel === channel);
if (pool == null) {
pool = new Pool(this, channel, this.genId());
this.sharedConnectionPools.push(pool);
}
const connection = new SharedConnection<Channels[C]>(
this,
channel,
pool,
name,
);
this.sharedConnections.push(connection as unknown as SharedConnection);
return connection;
}
public removeSharedConnection(connection: SharedConnection): void {
this.sharedConnections = this.sharedConnections.filter(
(c) => c !== connection,
);
}
public removeSharedConnectionPool(pool: Pool): void {
this.sharedConnectionPools = this.sharedConnectionPools.filter(
(p) => p !== pool,
);
}
private connectToChannel<C extends keyof Channels>(
channel: C,
params: Channels[C]["params"],
): NonSharedConnection<Channels[C]> {
const connection = new NonSharedConnection(
this,
channel,
this.genId(),
params,
);
this.nonSharedConnections.push(
connection as unknown as NonSharedConnection,
);
return connection;
}
public disconnectToChannel(connection: NonSharedConnection): void {
this.nonSharedConnections = this.nonSharedConnections.filter(
(c) => c !== connection,
);
}
/**
* Callback of when open connection
*/
private onOpen(): void {
const isReconnect = this.state === "reconnecting";
this.state = "connected";
this.emit("_connected_");
// チャンネル再接続
if (isReconnect) {
for (const p of this.sharedConnectionPools) p.connect();
for (const c of this.nonSharedConnections) c.connect();
}
}
/**
* Callback of when close connection
*/
private onClose(): void {
if (this.state === "connected") {
this.state = "reconnecting";
this.emit("_disconnected_");
}
}
/**
* Callback of when received a message from connection
*/
private onMessage(message: { data: string }): void {
const { type, body } = JSON.parse(message.data);
if (type === "channel") {
const id = body.id;
let connections: Connection[];
connections = this.sharedConnections.filter((c) => c.id === id);
if (connections.length === 0) {
const found = this.nonSharedConnections.find((c) => c.id === id);
if (found) {
connections = [found];
}
}
for (const c of connections) {
c.emit(body.type, body.body);
c.inCount++;
}
} else {
this.emit(type, body);
}
}
/**
* Send a message to connection
* ! ストリーム上のやり取りはすべてJSONで行われます !
*/
public send(typeOrPayload: string): void;
public send(typeOrPayload: string, payload: unknown): void;
public send(typeOrPayload: Record<string, unknown> | unknown[]): void;
public send(
typeOrPayload: string | Record<string, unknown> | unknown[],
payload?: unknown,
): void {
if (typeof typeOrPayload === "string") {
this.stream.send(
JSON.stringify({
type: typeOrPayload,
...(payload === undefined ? {} : { body: payload }),
}),
);
return;
}
this.stream.send(JSON.stringify(typeOrPayload));
}
public ping(): void {
this.stream.send("ping");
}
public heartbeat(): void {
this.stream.send("h");
}
/**
* Close this connection
*/
public close(): void {
this.stream.close();
}
}
// TODO: これらのクラスを Stream クラスの内部クラスにすれば余計なメンバをpublicにしないで済むかも
// もしくは @internal を使う? https://www.typescriptlang.org/tsconfig#stripInternal
class Pool {
public channel: string;
public id: string;
protected stream: Stream;
public users = 0;
private disposeTimerId: ReturnType<typeof setTimeout> | null = null;
private isConnected = false;
constructor(stream: Stream, channel: string, id: string) {
this.onStreamDisconnected = this.onStreamDisconnected.bind(this);
this.inc = this.inc.bind(this);
this.dec = this.dec.bind(this);
this.connect = this.connect.bind(this);
this.disconnect = this.disconnect.bind(this);
this.channel = channel;
this.stream = stream;
this.id = id;
this.stream.on("_disconnected_", this.onStreamDisconnected);
}
private onStreamDisconnected(): void {
this.isConnected = false;
}
public inc(): void {
if (this.users === 0 && !this.isConnected) {
this.connect();
}
this.users++;
// タイマー解除
if (this.disposeTimerId) {
clearTimeout(this.disposeTimerId);
this.disposeTimerId = null;
}
}
public dec(): void {
this.users--;
// そのコネクションの利用者が誰もいなくなったら
if (this.users === 0) {
// また直ぐに再利用される可能性があるので、一定時間待ち、
// 新たな利用者が現れなければコネクションを切断する
this.disposeTimerId = setTimeout(() => {
this.disconnect();
}, 3000);
}
}
public connect(): void {
if (this.isConnected) return;
this.isConnected = true;
this.stream.send("connect", {
channel: this.channel,
id: this.id,
});
}
private disconnect(): void {
this.stream.off("_disconnected_", this.onStreamDisconnected);
this.stream.send("disconnect", { id: this.id });
this.stream.removeSharedConnectionPool(this);
}
}
export interface IChannelConnection<
Channel extends AnyOf<Channels> = AnyOf<Channels>,
> extends EventEmitter<Channel["events"]> {
id: string;
name?: string;
inCount: number;
outCount: number;
channel: string;
send<T extends keyof Channel["receives"]>(
type: T,
body: Channel["receives"][T],
): void;
dispose(): void;
}
export abstract class Connection<
Channel extends AnyOf<Channels> = AnyOf<Channels>,
>
extends EventEmitter<Channel["events"]>
implements IChannelConnection<Channel>
{
public channel: string;
protected stream: Stream;
public abstract id: string;
public name?: string; // for debug
public inCount = 0; // for debug
public outCount = 0; // for debug
constructor(stream: Stream, channel: string, name?: string) {
super();
this.send = this.send.bind(this);
this.stream = stream;
this.channel = channel;
if (name !== undefined) {
this.name = name;
}
}
public send<T extends keyof Channel["receives"]>(
type: T,
body: Channel["receives"][T],
): void {
this.stream.send("ch", {
id: this.id,
type: type,
body: body,
});
this.outCount++;
}
public abstract dispose(): void;
}
class SharedConnection<
Channel extends AnyOf<Channels> = AnyOf<Channels>,
> extends Connection<Channel> {
private pool: Pool;
public get id(): string {
return this.pool.id;
}
constructor(stream: Stream, channel: string, pool: Pool, name?: string) {
super(stream, channel, name);
this.dispose = this.dispose.bind(this);
this.pool = pool;
this.pool.inc();
}
public dispose(): void {
this.pool.dec();
this.removeAllListeners();
this.stream.removeSharedConnection(this as unknown as SharedConnection);
}
}
class NonSharedConnection<
Channel extends AnyOf<Channels> = AnyOf<Channels>,
> extends Connection<Channel> {
public id: string;
protected params: Channel["params"];
constructor(
stream: Stream,
channel: string,
id: string,
params: Channel["params"],
) {
super(stream, channel);
this.connect = this.connect.bind(this);
this.dispose = this.dispose.bind(this);
this.params = params;
this.id = id;
this.connect();
}
public connect(): void {
this.stream.send("connect", {
channel: this.channel,
id: this.id,
params: this.params,
});
}
public dispose(): void {
this.removeAllListeners();
this.stream.send("disconnect", { id: this.id });
this.stream.disconnectToChannel(this as unknown as NonSharedConnection);
}
}

View File

@@ -0,0 +1,279 @@
import type {
Antenna,
DriveFile,
DriveFolder,
Note,
Notification,
Signin,
User,
UserDetailed,
UserDetailedNotMe,
UserLite,
} from "./autogen/models.js";
import type { ReversiUpdateKey } from "./consts.js";
import type {
AnnouncementCreated,
EmojiAdded,
EmojiDeleted,
EmojiUpdated,
PageEvent,
QueueStats,
QueueStatsLog,
ReversiGameDetailed,
ServerStats,
ServerStatsLog,
} from "./entities.js";
type ReversiUpdateSettings<K extends ReversiUpdateKey> = {
key: K;
value: ReversiGameDetailed[K];
};
export type Channels = {
main: {
params: null;
events: {
notification: (payload: Notification) => void;
mention: (payload: Note) => void;
reply: (payload: Note) => void;
renote: (payload: Note) => void;
follow: (payload: UserDetailedNotMe) => void; // 自分が他人をフォローしたとき
followed: (payload: UserDetailed | UserLite) => void; // 他人が自分をフォローしたとき
unfollow: (payload: UserDetailed) => void; // 自分が他人をフォロー解除したとき
meUpdated: (payload: UserDetailed) => void;
pageEvent: (payload: PageEvent) => void;
urlUploadFinished: (payload: { marker: string; file: DriveFile }) => void;
readAllNotifications: () => void;
unreadNotification: (payload: Notification) => void;
unreadMention: (payload: Note["id"]) => void;
readAllUnreadMentions: () => void;
notificationFlushed: () => void;
unreadSpecifiedNote: (payload: Note["id"]) => void;
readAllUnreadSpecifiedNotes: () => void;
readAllAntennas: () => void;
unreadAntenna: (payload: Antenna) => void;
readAllAnnouncements: () => void;
myTokenRegenerated: () => void;
signin: (payload: Signin) => void;
registryUpdated: (payload: {
scope?: string[];
key: string;
// biome-ignore lint/suspicious/noExplicitAny: <explanation>
value: any | null;
}) => void;
driveFileCreated: (payload: DriveFile) => void;
readAntenna: (payload: Antenna) => void;
receiveFollowRequest: (payload: User) => void;
announcementCreated: (payload: AnnouncementCreated) => void;
};
receives: null;
};
homeTimeline: {
params: {
withRenotes?: boolean;
withFiles?: boolean;
};
events: {
note: (payload: Note) => void;
};
receives: null;
};
localTimeline: {
params: {
withRenotes?: boolean;
withReplies?: boolean;
withFiles?: boolean;
};
events: {
note: (payload: Note) => void;
};
receives: null;
};
hybridTimeline: {
params: {
withRenotes?: boolean;
withReplies?: boolean;
withFiles?: boolean;
};
events: {
note: (payload: Note) => void;
};
receives: null;
};
globalTimeline: {
params: {
withRenotes?: boolean;
withFiles?: boolean;
};
events: {
note: (payload: Note) => void;
};
receives: null;
};
userList: {
params: {
listId: string;
withFiles?: boolean;
withRenotes?: boolean;
};
events: {
note: (payload: Note) => void;
};
receives: null;
};
hashtag: {
params: {
q: string[][];
};
events: {
note: (payload: Note) => void;
};
receives: null;
};
roleTimeline: {
params: {
roleId: string;
};
events: {
note: (payload: Note) => void;
};
receives: null;
};
antenna: {
params: {
antennaId: string;
};
events: {
note: (payload: Note) => void;
};
receives: null;
};
channel: {
params: {
channelId: string;
};
events: {
note: (payload: Note) => void;
};
receives: null;
};
drive: {
params: null;
events: {
fileCreated: (payload: DriveFile) => void;
fileDeleted: (payload: DriveFile["id"]) => void;
fileUpdated: (payload: DriveFile) => void;
folderCreated: (payload: DriveFolder) => void;
folderDeleted: (payload: DriveFolder["id"]) => void;
folderUpdated: (payload: DriveFolder) => void;
};
receives: null;
};
serverStats: {
params: null;
events: {
stats: (payload: ServerStats) => void;
statsLog: (payload: ServerStatsLog) => void;
};
receives: {
requestLog: {
id: string | number;
length: number;
};
};
};
queueStats: {
params: null;
events: {
stats: (payload: QueueStats) => void;
statsLog: (payload: QueueStatsLog) => void;
};
receives: {
requestLog: {
id: string | number;
length: number;
};
};
};
admin: {
params: null;
events: {
newAbuseUserReport: {
id: string;
targetUserId: string;
reporterId: string;
comment: string;
};
};
receives: null;
};
reversiGame: {
params: {
gameId: string;
};
events: {
started: (payload: { game: ReversiGameDetailed }) => void;
ended: (payload: {
winnerId: User["id"] | null;
game: ReversiGameDetailed;
}) => void;
canceled: (payload: { userId: User["id"] }) => void;
changeReadyStates: (payload: { user1: boolean; user2: boolean }) => void;
updateSettings: <K extends ReversiUpdateKey>(payload: {
userId: User["id"];
key: K;
value: ReversiGameDetailed[K];
}) => void;
log: (payload: Record<string, unknown>) => void;
};
receives: {
putStone: {
pos: number;
id: string;
};
ready: boolean;
cancel: null | Record<string, never>;
updateSettings: ReversiUpdateSettings<ReversiUpdateKey>;
claimTimeIsUp: null | Record<string, never>;
};
};
};
export type NoteUpdatedEvent = { id: Note["id"] } & (
| {
type: "reacted";
body: {
reaction: string;
emoji: string | null;
userId: User["id"];
};
}
| {
type: "unreacted";
body: {
reaction: string;
userId: User["id"];
};
}
| {
type: "deleted";
body: {
deletedAt: string;
};
}
| {
type: "pollVoted";
body: {
choice: number;
userId: User["id"];
};
}
);
export type BroadcastEvents = {
noteUpdated: (payload: NoteUpdatedEvent) => void;
emojiAdded: (payload: EmojiAdded) => void;
emojiUpdated: (payload: EmojiUpdated) => void;
emojiDeleted: (payload: EmojiDeleted) => void;
announcementCreated: (payload: AnnouncementCreated) => void;
};