| 1 | /** |
| 2 | * One answer from @g1t, from asking to its last word: what the channel's |
| 3 | * room (src/room.ts) runs for a question in a direct message with g1t. |
| 4 | * Everything it touches comes in as `AnswerDeps`, so it is tested without |
| 5 | * a provider, a database or sockets (answer.test.ts). |
| 6 | * |
| 7 | * 1. Asks the integrations service what Ask answers with. Nothing chosen: |
| 8 | * g1t says where an owner chooses a provider, and that is the answer. |
| 9 | * Chosen but unusable: g1t says why. |
| 10 | * 2. Asks, and reads the answer as it streams: the text so far goes to the |
| 11 | * room live at most every `SHOW_EVERY_MS`, and is saved every |
| 12 | * `SAVE_EVERY_MS`, so a reload mid-answer shows most of it. |
| 13 | * 3. Ends it: the message is stored as a normal message with what came |
| 14 | * (or a short line saying why nothing did), and the person is notified. |
| 15 | * |
| 16 | * Stop, or a newer question, aborts `signal`: what came stays, marked |
| 17 | * stopped. |
| 18 | */ |
| 19 | import type { AskSetup, ChatAnswer } from "@g1t/contracts"; |
| 20 | |
| 21 | import { AnswerReader, cannotAnswerText, finalBody, notSetUpText } from "./g1t.ts"; |
| 22 | |
| 23 | /** The shortest time between two live frames of an answer. */ |
| 24 | export const SHOW_EVERY_MS = 120; |
| 25 | /** The shortest time between two saves of an answer still coming. */ |
| 26 | export const SAVE_EVERY_MS = 3_000; |
| 27 | |
| 28 | export type AnswerDeps = { |
| 29 | slug: string; |
| 30 | /** What Ask answers with; null when the integrations service did not say. */ |
| 31 | setup(): Promise<AskSetup | null>; |
| 32 | /** The integrations service's `ask` answer, streamed. */ |
| 33 | ask(signal: AbortSignal): Promise<Response>; |
| 34 | /** g1t is thinking: shown as typing. */ |
| 35 | typing(): void; |
| 36 | /** The answer so far, live. */ |
| 37 | show(body: string, answer: ChatAnswer): void; |
| 38 | /** Saves the answer so far, while it is still coming. */ |
| 39 | save(body: string, answer: ChatAnswer): Promise<void>; |
| 40 | /** Stores the answer as it ended and tells the room and the person. */ |
| 41 | finish(body: string, answer: ChatAnswer): Promise<void>; |
| 42 | now(): number; |
| 43 | }; |
| 44 | |
| 45 | export async function runAnswer(deps: AnswerDeps, signal: AbortSignal): Promise<void> { |
| 46 | deps.typing(); |
| 47 | const setup = await deps.setup().catch(() => null); |
| 48 | if (signal.aborted) { |
| 49 | await deps.finish(finalBody("", { state: "stopped" }), { state: "stopped" }); |
| 50 | return; |
| 51 | } |
| 52 | if (setup?.problem) { |
| 53 | if (!setup.connection_id) { |
| 54 | await deps.finish(notSetUpText(deps.slug), { state: "done", setup: "missing" }); |
| 55 | } else { |
| 56 | await deps.finish(cannotAnswerText(setup.problem), { state: "error", error: setup.problem }); |
| 57 | } |
| 58 | return; |
| 59 | } |
| 60 | const reader = new AnswerReader(); |
| 61 | let shownAt = 0; |
| 62 | let savedAt = deps.now(); |
| 63 | try { |
| 64 | const response = await deps.ask(signal); |
| 65 | if (!response.ok || !response.body) { |
| 66 | reader.feed(`${JSON.stringify({ type: "error", message: "g1t's integrations did not answer. Try again in a moment." })}\n`); |
| 67 | } else { |
| 68 | const chunks = response.body.pipeThrough(new TextDecoderStream()).getReader(); |
| 69 | // Stop ends the read at once, not at the provider's next word. |
| 70 | const stopped = () => void chunks.cancel().catch(() => {}); |
| 71 | signal.addEventListener("abort", stopped, { once: true }); |
| 72 | try { |
| 73 | for (;;) { |
| 74 | const { value, done } = await chunks.read(); |
| 75 | if (done || signal.aborted) break; |
| 76 | if (!reader.feed(value)) continue; |
| 77 | const now = deps.now(); |
| 78 | if (now - shownAt >= SHOW_EVERY_MS) { |
| 79 | shownAt = now; |
| 80 | deps.show(reader.body, reader.answer); |
| 81 | } |
| 82 | if (now - savedAt >= SAVE_EVERY_MS && !reader.over) { |
| 83 | savedAt = now; |
| 84 | await deps.save(reader.body, reader.answer).catch(() => {}); |
| 85 | } |
| 86 | if (reader.over) break; |
| 87 | } |
| 88 | } finally { |
| 89 | signal.removeEventListener("abort", stopped); |
| 90 | } |
| 91 | } |
| 92 | } catch (error) { |
| 93 | if (!signal.aborted) { |
| 94 | console.error("g1t's answer broke off", error); |
| 95 | reader.feed(`${JSON.stringify({ type: "error", message: "The answer broke off. Ask again." })}\n`); |
| 96 | } |
| 97 | } |
| 98 | if (signal.aborted) reader.stop(); |
| 99 | else reader.finish(); |
| 100 | await deps.finish(finalBody(reader.body, reader.answer), reader.answer); |
| 101 | } |