cannorin/arubinochan-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Biome fix

Browse Source
This commit is contained in:
cannorin
2025-01-12 15:01:10 +00:00
parent c43a79a134
commit e9ee87920c
4 changed files with 444 additions and 356 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
index.ts
View File

@@ -1,7 +1,7 @@
import { api } from "misskey-js"; import { api } from "misskey-js";
import Stream from "./misskey-js/streaming";
import type { Stream as MisskeyStream } from "misskey-js"; import type { Stream as MisskeyStream } from "misskey-js";
import type { Note, UserLite } from "misskey-js/entities.js"; import type { Note, UserLite } from "misskey-js/entities.js";
import Stream from "./misskey-js/streaming";
import OpenAI from "openai"; import OpenAI from "openai";
import type { ChatCompletionMessageParam } from "openai/resources/index.js"; import type { ChatCompletionMessageParam } from "openai/resources/index.js";
@@ -16,7 +16,6 @@ const openai = new OpenAI({
apiKey: Bun.env["OPENAI_API_KEY"], apiKey: Bun.env["OPENAI_API_KEY"],
}); });
// #region util // #region util
/** pick up to random N elements from array. /** pick up to random N elements from array.
@@ -37,42 +36,40 @@ const sleep = (msec: number) =>
new Promise((resolve) => setTimeout(resolve, msec)); new Promise((resolve) => setTimeout(resolve, msec));
// #endregion // #endregion
// #region misskey // #region misskey
const me = await misskey.request("i", {}); const me = await misskey.request("i", {});
/** check if a note is suitable as an input */ /** check if a note is suitable as an input */
const isSuitableAsInput = (n: Note) => const isSuitableAsInput = (n: Note) =>
!n.user.isBot !n.user.isBot &&
&& !n.replyId !n.replyId &&
&& (!n.mentions || n.mentions.length === 0) (!n.mentions || n.mentions.length === 0) &&
&& n.text?.length && n.text.length > 0 n.text?.length &&
n.text.length > 0;
/** randomly sample some notes from the timeline */ /** randomly sample some notes from the timeline */
async function getNotes() { async function getNotes() {
// randomly sample N local notes // randomly sample N local notes
const localNotes = (count: number) => const localNotes = (count: number) =>
misskey.request("notes/local-timeline", { limit: 100 }) misskey
.request("notes/local-timeline", { limit: 100 })
.then((xs) => xs.filter(isSuitableAsInput)) .then((xs) => xs.filter(isSuitableAsInput))
.then((xs) => sample(xs, count)); .then((xs) => sample(xs, count));
// randomly sample N global notes // randomly sample N global notes
const globalNotes = (count: number) => const globalNotes = (count: number) =>
misskey.request("notes/global-timeline", { limit: 100 }) misskey
.request("notes/global-timeline", { limit: 100 })
.then((xs) => xs.filter(isSuitableAsInput)) .then((xs) => xs.filter(isSuitableAsInput))
.then((xs) => sample(xs, count)); .then((xs) => sample(xs, count));
// randomly sample N notes of mine // randomly sample N notes of mine
const myNotes = (count: number) => const myNotes = (count: number) =>
misskey.request("users/notes", { userId: me.id, limit: 100 }) misskey
.request("users/notes", { userId: me.id, limit: 100 })
.then((xs) => sample(xs, count)); .then((xs) => sample(xs, count));
const notes = await Promise.all([ const notes = await Promise.all([localNotes(5), globalNotes(10), myNotes(2)]);
localNotes(5),
globalNotes(10),
myNotes(2),
]);
return sample(notes.flat()); return sample(notes.flat());
} }
@@ -87,16 +84,20 @@ async function expandReplyTree(note: Note, acc: Note[] = [], cutoff = 5) {
const noteToMessage = (note: Note): ChatCompletionMessageParam => ({ const noteToMessage = (note: Note): ChatCompletionMessageParam => ({
role: note.userId === me.id ? ("assistant" as const) : ("user" as const), role: note.userId === me.id ? ("assistant" as const) : ("user" as const),
content: note.text?.replaceAll(`@${me.username}`, "") || "", content: note.text?.replaceAll(`@${me.username}`, "") || "",
}) });
// #endregion // #endregion
// #region job // #region job
type Job = type Job =
// read posts and post a note // read posts and post a note
| { type: "post" } | { type: "post" }
// reply to a specific note // reply to a specific note
| { type: "reply"; id: string, visibility: Note["visibility"], replyTree: Note[] }; | {
type: "reply";
id: string;
visibility: Note["visibility"];
replyTree: Note[];
};
/** create a prompt for the job */ /** create a prompt for the job */
async function preparePrompt(job: Job): Promise<ChatCompletionMessageParam[]> { async function preparePrompt(job: Job): Promise<ChatCompletionMessageParam[]> {
@@ -154,7 +155,7 @@ user が SNS 上で、あなたへのメッセージを送ってきています
このような文体を真似して、user の発言に返答してください。`, このような文体を真似して、user の発言に返答してください。`,
}, },
...(job.replyTree.map(noteToMessage)) ...job.replyTree.map(noteToMessage),
]; ];
} }
} }
@@ -187,7 +188,8 @@ async function processJob(job: Job) {
console.log(); console.log();
// concatenate the partial responses // concatenate the partial responses
const text = responses.join("") const text = responses
.join("")
.replaceAll(/(\r\n|\r|\n)\s+/g, "\n\n") // remove extra newlines .replaceAll(/(\r\n|\r|\n)\s+/g, "\n\n") // remove extra newlines
.replaceAll("@", ""); // remove mentions .replaceAll("@", ""); // remove mentions
@@ -221,8 +223,8 @@ function initializeStream() {
token: Bun.env["MISSKEY_CREDENTIAL"] ?? "", token: Bun.env["MISSKEY_CREDENTIAL"] ?? "",
}, },
{ {
binaryType: "arraybuffer" binaryType: "arraybuffer",
} },
) as unknown as MisskeyStream; ) as unknown as MisskeyStream;
channel = stream.useChannel("main"); channel = stream.useChannel("main");
@@ -241,7 +243,12 @@ function initializeStream() {
if (e.text && e.userId !== me.id && !e.user.isBot) { if (e.text && e.userId !== me.id && !e.user.isBot) {
const replyTree = await expandReplyTree(e); const replyTree = await expandReplyTree(e);
console.log(`* push: reply (${e.id}, ${replyTree.length} msgs)`); console.log(`* push: reply (${e.id}, ${replyTree.length} msgs)`);
jobs.push({ type: "reply", id: e.id, visibility: e.visibility, replyTree }); jobs.push({
type: "reply",
id: e.id,
visibility: e.visibility,
replyTree,
});
} }
}); });
@@ -276,7 +283,10 @@ async function pushJob() {
while (true) { while (true) {
const now = new Date(Date.now()); const now = new Date(Date.now());
// push a post job every 15 minutes (XX:00, XX:15, XX:30, XX:45) // push a post job every 15 minutes (XX:00, XX:15, XX:30, XX:45)
if (now.getMinutes() % 15 < Number.EPSILON && !jobs.some((job) => job.type === "post")) { if (
now.getMinutes() % 15 < Number.EPSILON &&
!jobs.some((job) => job.type === "post")
) {
console.log("* push: post"); console.log("* push: post");
jobs.push({ type: "post" }); jobs.push({ type: "post" });
} }

685
misskey-js/streaming.ts
View File

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

48
misskey-js/streaming.types.ts
View File

@@ -241,33 +241,33 @@ export type Channels = {
export type NoteUpdatedEvent = { id: Note["id"] } & ( export type NoteUpdatedEvent = { id: Note["id"] } & (
| { | {
type: "reacted"; type: "reacted";
body: { body: {
reaction: string; reaction: string;
emoji: string | null; emoji: string | null;
userId: User["id"]; userId: User["id"];
}; };
} }
| { | {
type: "unreacted"; type: "unreacted";
body: { body: {
reaction: string; reaction: string;
userId: User["id"]; userId: User["id"];
}; };
} }
| { | {
type: "deleted"; type: "deleted";
body: { body: {
deletedAt: string; deletedAt: string;
}; };
} }
| { | {
type: "pollVoted"; type: "pollVoted";
body: { body: {
choice: number; choice: number;
userId: User["id"]; userId: User["id"];
}; };
} }
); );
export type BroadcastEvents = { export type BroadcastEvents = {

7
tsconfig.json
View File

@@ -2,10 +2,7 @@
"extends": "@tsconfig/strictest/tsconfig.json", "extends": "@tsconfig/strictest/tsconfig.json",
"compilerOptions": { "compilerOptions": {
// Enable latest features // Enable latest features
"lib": [ "lib": ["ESNext", "DOM"],
"ESNext",
"DOM"
],
"target": "ESNext", "target": "ESNext",
"module": "ESNext", "module": "ESNext",
"moduleDetection": "force", "moduleDetection": "force",
@@ -17,4 +14,4 @@
"verbatimModuleSyntax": true, "verbatimModuleSyntax": true,
"noEmit": true "noEmit": true
} }
} }