refactor(BREAKING): remove listenStream and add listen that add event listners for socket.io

Author: takker99

browser/websocket/applyCommit.ts

@@ -1,4 +1,4 @@
-import type { CommitNotification } from "./websocket-types.ts";
+import type { CommitNotification } from "./listen-events.ts";
 import type { BaseLine } from "@cosense/types/rest";
 import { getUnixTimeFromId } from "./id.ts";
 

browser/websocket/change.ts

@@ -0,0 +1,68 @@
+export type Change =
+  | InsertChange
+  | UpdateChange
+  | DeleteChange
+  | LinksChange
+  | ProjectLinksChange
+  | IconsChange
+  | DescriptionsChange
+  | ImageChange
+  | FilesChange
+  | HelpFeelsChange
+  | infoboxDefinitionChange
+  | TitleChange;
+export interface InsertChange {
+  _insert: string;
+  lines: {
+    id: string;
+    text: string;
+  };
+}
+export interface UpdateChange {
+  _update: string;
+  lines: {
+    text: string;
+  };
+  noTimestampUpdate?: unknown;
+}
+export interface DeleteChange {
+  _delete: string;
+  lines: -1;
+}
+export interface LinksChange {
+  links: string[];
+}
+export interface ProjectLinksChange {
+  projectLinks: string[];
+}
+export interface IconsChange {
+  icons: string[];
+}
+export interface DescriptionsChange {
+  descriptions: string[];
+}
+export interface ImageChange {
+  image: string | null;
+}
+export interface TitleChange {
+  title: string;
+}
+export interface FilesChange {
+  /** file id */
+  files: string[];
+}
+export interface HelpFeelsChange {
+  /** Helpfeel記法の先頭の`? `をとったもの */
+  helpfeels: string[];
+}
+export interface infoboxDefinitionChange {
+  /** `table:infobox`または`table:cosense`の各行をtrimしたもの */
+  infoboxDefinition: string[];
+}
+export interface PinChange {
+  pin: number;
+}
+export interface DeletePageChange {
+  deleted: true;
+  merged?: true;
+}

browser/websocket/diffToChanges.ts

@@ -1,10 +1,6 @@
 import { diff, toExtendedChanges } from "../../deps/onp.ts";
 import type { Line } from "@cosense/types/userscript";
-import type {
-  DeleteChange,
-  InsertChange,
-  UpdateChange,
-} from "./websocket-types.ts";
+import type { DeleteChange, InsertChange, UpdateChange } from "./change.ts";
 import { createNewLineId } from "./id.ts";
 
 type Options = {

browser/websocket/emit-events.ts

@@ -0,0 +1,75 @@
+import type { Change, DeletePageChange, PinChange } from "./change.ts";
+
+export interface EmitEvents {
+  "socket.io-request": (
+    req: { method: "commit"; data: PageCommit } | {
+      method: "room:join";
+      data: JoinRoomRequest;
+    },
+    callback: (
+      res:
+        | { data: PageCommitResponse | JoinRoomResponse }
+        | { error: { name: string; message?: string } },
+    ) => void,
+  ) => void;
+  cursor: (req: Omit<MoveCursorData, "socketId">) => void;
+}
+
+export interface PageCommit {
+  kind: "page";
+  parentId: string;
+  projectId: string;
+  pageId: string;
+  userId: string;
+  changes: Change[] | [PinChange] | [DeletePageChange];
+  cursor?: null;
+  freeze: true;
+}
+
+export interface PageCommitResponse {
+  commitId: string;
+}
+
+export type JoinRoomRequest =
+  | JoinPageRoomRequest
+  | JoinProjectRoomRequest
+  | JoinStreamRoomRequest;
+
+export interface JoinProjectRoomRequest {
+  pageId: null;
+  projectId: string;
+  projectUpdatesStream: false;
+}
+
+export interface JoinPageRoomRequest {
+  pageId: string;
+  projectId: string;
+  projectUpdatesStream: false;
+}
+
+export interface JoinStreamRoomRequest {
+  pageId: null;
+  projectId: string;
+  projectUpdatesStream: true;
+}
+
+export interface JoinRoomResponse {
+  success: true;
+  pageId: string | null;
+  projectId: string;
+}
+
+export interface MoveCursorData {
+  user: {
+    id: string;
+    name: string;
+    displayName: string;
+  };
+  pageId: string;
+  position: {
+    line: number;
+    char: number;
+  };
+  visible: boolean;
+  socketId: string;
+}

browser/websocket/emit.ts

@@ -1,22 +1,20 @@
 import { createErr, createOk, type Result } from "option-t/plain_result";
 import type { Socket } from "socket.io-client";
 import type {
-  EmitEvents,
   JoinRoomRequest,
-  ListenEvents,
   MoveCursorData,
   PageCommit,
   PageCommitResponse,
-} from "./websocket-types.ts";
-export * from "./websocket-types.ts";
+} from "./emit-events.ts";
 import {
   isPageCommitError,
   type PageCommitError,
   type SocketIOServerDisconnectError,
   type TimeoutError,
   type UnexpectedRequestError,
 } from "./error.ts";
-import type { JoinRoomResponse } from "./websocket-types.ts";
+import type { JoinRoomResponse } from "./emit-events.ts";
+import type { ScrapboxSocket } from "./socket.ts";
 
 export interface WrapperdEmitEvents {
   commit: { req: PageCommit; res: PageCommitResponse; err: PageCommitError };
@@ -47,7 +45,7 @@ export interface EmitOptions {
  * @returns  A promise that resolves with the result of the emit operation.
  */
 export const emit = <EventName extends keyof WrapperdEmitEvents>(
-  socket: Socket<ListenEvents, EmitEvents>,
+  socket: ScrapboxSocket,
   event: EventName,
   data: WrapperdEmitEvents[EventName]["req"],
   options?: EmitOptions,

browser/websocket/listen-events.ts

@@ -0,0 +1,106 @@
+import type { MoveCursorData, PageCommit } from "./emit-events.ts";
+import type {
+  DeleteChange,
+  DeletePageChange,
+  DescriptionsChange,
+  IconsChange,
+  ImageChange,
+  InsertChange,
+  LinksChange,
+  TitleChange,
+  UpdateChange,
+} from "./change.ts";
+
+export interface ListenEvents {
+  "projectUpdatesStream:commit": (event: ProjectUpdatesStreamCommit) => void;
+  "projectUpdatesStream:event": (event: ProjectUpdatesStreamEvent) => void;
+  commit: (event: CommitNotification) => void;
+  cursor: (event: MoveCursorData) => void;
+  "quick-search:commit": (event: QuickSearchCommit) => void;
+  "quick-search:replace-link": QuickSearchReplaceLink;
+  "infobox:updating": boolean;
+  "infobox:reload": void;
+  "literal-database:reload": void;
+}
+
+export interface ProjectUpdatesStreamCommit {
+  kind: "page";
+  id: string;
+  parentId: string;
+  projectId: string;
+  pageId: string;
+  userId: string;
+  changes:
+    | (
+      | InsertChange
+      | UpdateChange
+      | DeleteChange
+      | TitleChange
+      | LinksChange
+      | IconsChange
+    )[]
+    | [DeletePageChange];
+  cursor: null;
+  freeze: true;
+}
+
+export type ProjectUpdatesStreamEvent =
+  | PageDeleteEvent
+  | MemberJoinEvent
+  | MemberAddEvent
+  | AdminAddEvent
+  | AdminDeleteEvent
+  | OwnerSetEvent
+  | InvitationResetEvent;
+
+export interface ProjectEvent {
+  id: string;
+  pageId: string;
+  userId: string;
+  projectId: string;
+  created: number;
+  updated: number;
+}
+
+export interface PageDeleteEvent extends ProjectEvent {
+  type: "page.delete";
+  data: {
+    titleLc: string;
+  };
+}
+export interface MemberJoinEvent extends ProjectEvent {
+  type: "member.join";
+}
+export interface MemberAddEvent extends ProjectEvent {
+  type: "member.add";
+}
+export interface InvitationResetEvent extends ProjectEvent {
+  type: "invitation.reset";
+}
+export interface AdminAddEvent extends ProjectEvent {
+  type: "admin.add";
+  targetUserId: string;
+}
+export interface AdminDeleteEvent extends ProjectEvent {
+  type: "admin.delete";
+  targetUserId: string;
+}
+export interface OwnerSetEvent extends ProjectEvent {
+  type: "owner.set";
+  targetUserId: string;
+}
+
+export interface CommitNotification extends PageCommit {
+  id: string;
+}
+
+export interface QuickSearchCommit extends Omit<CommitNotification, "changes"> {
+  changes:
+    | (TitleChange | LinksChange | DescriptionsChange | ImageChange)[]
+    | [DeletePageChange];
+}
+
+export interface QuickSearchReplaceLink {
+  from: string;
+  to: string;
+}

browser/websocket/listen.ts

@@ -1,26 +1,21 @@
-import { createOk, isErr, type Result, unwrapOk } from "option-t/plain_result";
 import type {
   NotFoundError,
   NotLoggedInError,
   NotMemberError,
 } from "@cosense/types/rest";
-import type {
-  ProjectUpdatesStreamCommit,
-  ProjectUpdatesStreamEvent,
-} from "./emit.ts";
 import type { HTTPError } from "../../rest/responseIntoResult.ts";
 import type { AbortError, NetworkError } from "../../rest/robustFetch.ts";
-import { getProjectId } from "./pull.ts";
-import { connect, disconnect } from "./socket.ts";
-import type { Socket } from "socket.io-client";
+import type { ScrapboxSocket } from "./socket.ts";
+import type { ListenEvents } from "./listen-events.ts";
 
 export type {
   ProjectUpdatesStreamCommit,
   ProjectUpdatesStreamEvent,
-} from "./websocket-types.ts";
+} from "./listen-events.ts";
 
 export interface ListenStreamOptions {
-  socket?: Socket;
+  signal?: AbortSignal;
+  once?: boolean;
 }
 
 export type ListenStreamError =
@@ -37,46 +32,21 @@ export type ListenStreamError =
  * @param events 購読したいevent。配列で指定する
  * @param options 使用したいSocketがあれば指定する
  */
-export async function* listenStream(
-  project: string,
-  events: ["commit" | "event", ...("commit" | "event")[]],
+export const listen = <EventName extends keyof ListenEvents>(
+  socket: ScrapboxSocket,
+  event: EventName,
+  listener: ListenEvents[EventName],
   options?: ListenStreamOptions,
-): AsyncGenerator<
-  Result<
-    ProjectUpdatesStreamEvent | ProjectUpdatesStreamCommit,
-    ListenStreamError
-  >,
-  void,
-  unknown
-> {
-  const result = await getProjectId(project);
-  if (isErr(result)) {
-    yield result;
-    return;
-  }
-  const projectId = unwrapOk(result);
+): void => {
+  if (options?.signal?.aborted) return;
 
-  const injectedSocket = options?.socket;
-  const result2 = await connect(injectedSocket);
-  if (isErr(result2)) throw new Error("Failed to connect to websocket");
-  const socket = unwrapOk(result2);
-  const { request, response } = wrap(socket);
+  // deno-lint-ignore no-explicit-any
+  (options?.once ? socket.once : socket.on)(event, listener as any);
 
-  try {
-    // 部屋に入って購読し始める
-    await request("socket.io-request", {
-      method: "room:join",
-      data: { projectId, pageId: null, projectUpdatesStream: true },
-    });
-
-    for await (
-      const streamEvent of response(
-        ...events.map((event) => `projectUpdatesStream:${event}` as const),
-      )
-    ) {
-      yield createOk(streamEvent);
-    }
-  } finally {
-    if (!injectedSocket) await disconnect(socket);
-  }
-}
+  options?.signal?.addEventListener?.(
+    "abort",
+    // deno-lint-ignore no-explicit-any
+    () => socket.off(event, listener as any),
+    { once: true },
+  );
+};