|
|
@@ -1,222 +1,81 @@
|
|
|
-import type { ChatInputCommandInteraction, GuildMember, Interaction, Message, TextChannel, ThreadChannel } from "discord.js"
|
|
|
+import type { ChatInputCommandInteraction, GuildMember, Interaction, Message, TextChannel } from "discord.js"
|
|
|
import { ChannelType, MessageFlags } from "discord.js"
|
|
|
import { Context, Effect, Layer, Option, Queue, Ref, Runtime, Schedule, Stream } from "effect"
|
|
|
-import { AppConfig } from "../../../config"
|
|
|
-import { DiscordClient } from "../../../discord/client"
|
|
|
-import { TYPING_INTERVAL } from "../../../discord/constants"
|
|
|
-import { cleanResponse, splitForDiscord } from "../../../discord/format"
|
|
|
-import { ThreadAgentPool } from "../../../sandbox/pool"
|
|
|
-import { SessionStore } from "../../../sessions/store"
|
|
|
-import { ChannelId, GuildId, ThreadId } from "../../../types"
|
|
|
-import { DeliveryError, HistoryError, messageOf, ThreadEnsureError } from "../../model/errors"
|
|
|
-import { ChannelMessage, Mention, ThreadMessage, ThreadRef, Typing, type Action, type Inbound } from "../../model/schema"
|
|
|
-import { ConversationLedger, History, Inbox, Outbox, Threads } from "../../services"
|
|
|
-
|
|
|
-type ChatChannel = TextChannel | ThreadChannel
|
|
|
-
|
|
|
-const HISTORY_FETCH_LIMIT = 40
|
|
|
-const HISTORY_LINE_CHAR_LIMIT = 500
|
|
|
-const HISTORY_TOTAL_CHAR_LIMIT = 6000
|
|
|
-const INGRESS_DEDUP_LIMIT = 4_000
|
|
|
-const EMPTY_MENTION_REPLY = "Tag me with a question!"
|
|
|
-const SETUP_FAILURE_REPLY = "Something went wrong setting up the thread."
|
|
|
-const COMMAND_NOT_THREAD_REPLY = "Use this command inside a Discord thread."
|
|
|
-const COMMAND_FORBIDDEN_REPLY = "You don't have the required role for this command."
|
|
|
-const COMMAND_CHANNEL_REPLY = "This thread is not allowed for the bot."
|
|
|
-const COMMAND_ACK = "Running command in this thread..."
|
|
|
+import { AppConfig } from "../config"
|
|
|
+import { DiscordClient } from "./client"
|
|
|
+import { TYPING_INTERVAL } from "./constants"
|
|
|
+import { cleanResponse, splitForDiscord } from "./format"
|
|
|
+import { SessionStore } from "../session/store"
|
|
|
+import { ChannelId, GuildId, ThreadId } from "../types"
|
|
|
+import { DeliveryError, HistoryError, messageOf, ThreadEnsureError } from "../conversation/model/errors"
|
|
|
+import { ChannelMessage, Mention, ThreadMessage, ThreadRef, Typing, type Action, type Inbound } from "../conversation/model/schema"
|
|
|
+import { History, Inbox, OffsetStore, Outbox, Threads } from "../conversation"
|
|
|
+import {
|
|
|
+ COMMAND_ACK,
|
|
|
+ COMMAND_CHANNEL_REPLY,
|
|
|
+ COMMAND_FORBIDDEN_REPLY,
|
|
|
+ COMMAND_NOT_THREAD_REPLY,
|
|
|
+ COMMANDS,
|
|
|
+ EMPTY_MENTION_REPLY,
|
|
|
+ SETUP_FAILURE_REPLY,
|
|
|
+ commandText,
|
|
|
+} from "./conversation-commands"
|
|
|
+import {
|
|
|
+ asTextChannel,
|
|
|
+ asThreadChannel,
|
|
|
+ type ChatChannel,
|
|
|
+ hasRequiredRole,
|
|
|
+ isChannelAllowed,
|
|
|
+ isMentioned,
|
|
|
+} from "./conversation-channels"
|
|
|
+import { catchupBenign, deliveryRetriable, deliveryRetry } from "./conversation-delivery"
|
|
|
+import { buildHistoryReplayPrompt } from "./conversation-history"
|
|
|
+import { catchupFromOffset } from "./catchup"
|
|
|
+
|
|
|
+const CACHE_LIMIT = 4_000
|
|
|
const CATCHUP_PAGE_SIZE = 100
|
|
|
-const COMMANDS = [
|
|
|
- {
|
|
|
- name: "status",
|
|
|
- description: "Show sandbox status for this thread",
|
|
|
- },
|
|
|
- {
|
|
|
- name: "reset",
|
|
|
- description: "Destroy the sandbox session for this thread",
|
|
|
- },
|
|
|
-] as const
|
|
|
-
|
|
|
-const commandText = (name: string): string => {
|
|
|
- if (name === "status") return "!status"
|
|
|
- if (name === "reset") return "!reset"
|
|
|
- return ""
|
|
|
-}
|
|
|
-
|
|
|
-const isChannelAllowed = (channelId: string, categoryId: string | null, config: AppConfig.Service): boolean => {
|
|
|
- if (config.allowedChannelIds.length > 0 && config.allowedChannelIds.includes(channelId)) return true
|
|
|
- if (config.discordCategoryId && categoryId === config.discordCategoryId) return true
|
|
|
- return false
|
|
|
-}
|
|
|
-
|
|
|
-const hasRequiredRole = (member: GuildMember | null, config: AppConfig.Service): boolean => {
|
|
|
- if (!config.discordRequiredRoleId) return true
|
|
|
- if (!member) return false
|
|
|
- return member.roles.cache.has(config.discordRequiredRoleId)
|
|
|
-}
|
|
|
-
|
|
|
-const asThreadChannel = (value: unknown): ThreadChannel | null => {
|
|
|
- if (typeof value !== "object" || value === null) return null
|
|
|
- const type = (value as { type?: unknown }).type
|
|
|
- if (type === ChannelType.PublicThread || type === ChannelType.PrivateThread) return value as ThreadChannel
|
|
|
- return null
|
|
|
-}
|
|
|
-
|
|
|
-const asTextChannel = (value: unknown): TextChannel | null => {
|
|
|
- if (typeof value !== "object" || value === null) return null
|
|
|
- const type = (value as { type?: unknown }).type
|
|
|
- if (type === ChannelType.GuildText) return value as TextChannel
|
|
|
- return null
|
|
|
-}
|
|
|
-
|
|
|
-const isMentioned = (message: Message, botUserId: string, botRoleId: string): boolean => {
|
|
|
- if (botUserId.length > 0 && message.mentions.users.has(botUserId)) return true
|
|
|
- if (botRoleId.length > 0 && message.mentions.roles.has(botRoleId)) return true
|
|
|
- if (botUserId.length > 0 && message.content.includes(`<@${botUserId}>`)) return true
|
|
|
- if (botUserId.length > 0 && message.content.includes(`<@!${botUserId}>`)) return true
|
|
|
- if (botRoleId.length > 0 && message.content.includes(`<@&${botRoleId}>`)) return true
|
|
|
- return false
|
|
|
-}
|
|
|
-
|
|
|
-const buildHistoryReplayPrompt = Effect.fn("DiscordAdapter.buildHistoryReplayPrompt")(
|
|
|
- function* (channel: ChatChannel, latest: string) {
|
|
|
- const fetched = yield* Effect.tryPromise(() => channel.messages.fetch({ limit: HISTORY_FETCH_LIMIT }))
|
|
|
- const ordered = [...fetched.values()].sort((a, b) => a.createdTimestamp - b.createdTimestamp)
|
|
|
- const lines = ordered
|
|
|
- .filter((prior) => !prior.system)
|
|
|
- .flatMap((prior) => {
|
|
|
- const text = prior.content.replace(/\s+/g, " ").trim()
|
|
|
- const files = prior.attachments.size > 0
|
|
|
- ? `[attachments: ${[...prior.attachments.values()].map((att) => att.name ?? "file").join(", ")}]`
|
|
|
- : ""
|
|
|
- const line = text || files
|
|
|
- if (!line) return []
|
|
|
- const value = line.length > HISTORY_LINE_CHAR_LIMIT ? `${line.slice(0, HISTORY_LINE_CHAR_LIMIT)}...` : line
|
|
|
- return [`${prior.author.bot ? "assistant" : "user"}: ${value}`]
|
|
|
- })
|
|
|
-
|
|
|
- const prior = lines.at(-1) === `user: ${latest}` ? lines.slice(0, -1) : lines
|
|
|
- if (prior.length === 0) return latest
|
|
|
-
|
|
|
- const selected = prior.reduceRight(
|
|
|
- (state, candidate) => {
|
|
|
- if (state.stop) return state
|
|
|
- if (state.total + candidate.length > HISTORY_TOTAL_CHAR_LIMIT && state.list.length > 0) {
|
|
|
- return { ...state, stop: true }
|
|
|
- }
|
|
|
- return { list: [candidate, ...state.list], total: state.total + candidate.length, stop: false }
|
|
|
- },
|
|
|
- { list: [] as ReadonlyArray<string>, total: 0, stop: false },
|
|
|
- ).list
|
|
|
-
|
|
|
- return [
|
|
|
- "Conversation history from this same Discord thread (oldest to newest):",
|
|
|
- selected.join("\n"),
|
|
|
- "",
|
|
|
- "Continue the same conversation and respond to the latest user message:",
|
|
|
- latest,
|
|
|
- ].join("\n")
|
|
|
- },
|
|
|
-)
|
|
|
-
|
|
|
-const statusOf = (cause: unknown): number | null => {
|
|
|
- if (typeof cause !== "object" || cause === null) return null
|
|
|
- const status = (cause as { status?: unknown }).status
|
|
|
- if (typeof status === "number") return status
|
|
|
- const code = (cause as { code?: unknown }).code
|
|
|
- if (typeof code === "number") return code
|
|
|
- return null
|
|
|
-}
|
|
|
-
|
|
|
-const deliveryRetriable = (cause: unknown): boolean => {
|
|
|
- const status = statusOf(cause)
|
|
|
- if (status === 429) return true
|
|
|
- if (status !== null && status >= 500) return true
|
|
|
- return false
|
|
|
-}
|
|
|
-
|
|
|
-const catchupBenign = (cause: unknown): boolean => {
|
|
|
- const text = messageOf(cause).toLowerCase()
|
|
|
- if (text.includes("missing access")) return true
|
|
|
- if (text.includes("missing permissions")) return true
|
|
|
- if (text.includes("unknown channel")) return true
|
|
|
- if (text.includes("50001")) return true
|
|
|
- if (text.includes("50013")) return true
|
|
|
- return false
|
|
|
-}
|
|
|
-
|
|
|
-const deliveryRetry = Schedule.exponential("200 millis").pipe(
|
|
|
- Schedule.intersect(Schedule.recurs(3)),
|
|
|
- Schedule.whileInput((error: DeliveryError) => error.retriable),
|
|
|
-)
|
|
|
|
|
|
export class DiscordConversationServices {
|
|
|
static readonly portLayer = Layer.scopedContext(
|
|
|
Effect.gen(function* () {
|
|
|
const client = yield* DiscordClient
|
|
|
const config = yield* AppConfig
|
|
|
- const pool = yield* ThreadAgentPool
|
|
|
const sessions = yield* SessionStore
|
|
|
- const ledger = yield* ConversationLedger
|
|
|
+ const offsets = yield* OffsetStore
|
|
|
const runtime = yield* Effect.runtime<never>()
|
|
|
const input = yield* Queue.unbounded<Inbound>()
|
|
|
const chats = new Map<string, ChatChannel>()
|
|
|
const texts = new Map<string, TextChannel>()
|
|
|
const refs = new Map<string, Message>()
|
|
|
const roots = new Map<string, ThreadId>()
|
|
|
- const seen = new Set<string>()
|
|
|
- const order: Array<string> = []
|
|
|
const ref_ids: Array<string> = []
|
|
|
const root_ids: Array<string> = []
|
|
|
|
|
|
- const mark = (message_id: string): boolean => {
|
|
|
- if (seen.has(message_id)) return false
|
|
|
- seen.add(message_id)
|
|
|
- order.push(message_id)
|
|
|
- if (order.length <= INGRESS_DEDUP_LIMIT) return true
|
|
|
- const oldest = order.shift()
|
|
|
- if (!oldest) return true
|
|
|
- seen.delete(oldest)
|
|
|
- return true
|
|
|
- }
|
|
|
-
|
|
|
const stash = <A>(map: Map<string, A>, keys: Array<string>, key: string, value: A) => {
|
|
|
if (!map.has(key)) keys.push(key)
|
|
|
map.set(key, value)
|
|
|
- if (keys.length <= INGRESS_DEDUP_LIMIT) return
|
|
|
+ if (keys.length <= CACHE_LIMIT) return
|
|
|
const oldest = keys.shift()
|
|
|
if (!oldest) return
|
|
|
map.delete(oldest)
|
|
|
}
|
|
|
|
|
|
- const sourceChannel = (channel_id: string) => `channel:${channel_id}`
|
|
|
- const sourceThread = (thread_id: string) => `thread:${thread_id}`
|
|
|
+ const sourceChannel = (channelId: string) => `channel:${channelId}`
|
|
|
+ const sourceThread = (threadId: string) => `thread:${threadId}`
|
|
|
const uniq = <A>(values: ReadonlyArray<A>): Array<A> => [...new Set(values)]
|
|
|
|
|
|
const offer = (event: Inbound, onFresh: Effect.Effect<void>) =>
|
|
|
- ledger.dedup(event.message_id).pipe(
|
|
|
- Effect.flatMap((fresh) => {
|
|
|
- if (!fresh) {
|
|
|
- return Effect.logDebug("Message deduped (already seen)").pipe(
|
|
|
- Effect.annotateLogs({
|
|
|
- event: "conversation.message.deduped",
|
|
|
- message_id: event.message_id,
|
|
|
- }),
|
|
|
- )
|
|
|
- }
|
|
|
- return Effect.logInfo("Message queued").pipe(
|
|
|
- Effect.annotateLogs({
|
|
|
- event: "conversation.message.queued",
|
|
|
- kind: event.kind,
|
|
|
- message_id: event.message_id,
|
|
|
- author_id: event.author_id,
|
|
|
- content: event.content.slice(0, 200),
|
|
|
- }),
|
|
|
- ).pipe(
|
|
|
- Effect.zipRight(onFresh),
|
|
|
- Effect.zipRight(input.offer(event)),
|
|
|
- Effect.asVoid,
|
|
|
- )
|
|
|
+ Effect.logInfo("Message queued").pipe(
|
|
|
+ Effect.annotateLogs({
|
|
|
+ event: "conversation.message.queued",
|
|
|
+ kind: event.kind,
|
|
|
+ message_id: event.messageId,
|
|
|
+ author_id: event.authorId,
|
|
|
+ content: event.content.slice(0, 200),
|
|
|
}),
|
|
|
+ Effect.zipRight(onFresh),
|
|
|
+ Effect.zipRight(input.offer(event)),
|
|
|
+ Effect.asVoid,
|
|
|
)
|
|
|
|
|
|
const memberOf = (message: Message) => {
|
|
|
@@ -238,102 +97,103 @@ export class DiscordConversationServices {
|
|
|
if (source === null) return
|
|
|
|
|
|
if (message.author.bot || message.mentions.everyone) {
|
|
|
- yield* ledger.setOffset(source, message.id)
|
|
|
+ yield* offsets.setOffset(source, message.id)
|
|
|
return
|
|
|
}
|
|
|
const member = yield* memberOf(message)
|
|
|
if (!hasRequiredRole(member, config)) {
|
|
|
- yield* ledger.setOffset(source, message.id)
|
|
|
+ yield* offsets.setOffset(source, message.id)
|
|
|
return
|
|
|
}
|
|
|
|
|
|
- const bot_user_id = client.user?.id ?? ""
|
|
|
- const bot_role_id = config.discordRoleId
|
|
|
- const mentioned = isMentioned(message, bot_user_id, bot_role_id)
|
|
|
+ const botUserId = client.user?.id ?? ""
|
|
|
+ const botRoleId = config.discordRoleId
|
|
|
+ const mentioned = isMentioned(message, botUserId, botRoleId)
|
|
|
const content = message.content.replace(/<@[!&]?\d+>/g, "").trim()
|
|
|
const mentions = Mention.make({
|
|
|
- user_ids: [...message.mentions.users.keys()],
|
|
|
- role_ids: [...message.mentions.roles.keys()],
|
|
|
+ userIds: [...message.mentions.users.keys()],
|
|
|
+ roleIds: [...message.mentions.roles.keys()],
|
|
|
})
|
|
|
|
|
|
if (!content && mentioned) {
|
|
|
yield* Effect.tryPromise(() => message.reply(EMPTY_MENTION_REPLY)).pipe(Effect.catchAll(() => Effect.void))
|
|
|
- yield* ledger.setOffset(source, message.id)
|
|
|
+ yield* offsets.setOffset(source, message.id)
|
|
|
return
|
|
|
}
|
|
|
|
|
|
if (message.channel.type === ChannelType.PublicThread || message.channel.type === ChannelType.PrivateThread) {
|
|
|
- const thread = message.channel as ThreadChannel
|
|
|
- const thread_id = ThreadId.make(thread.id)
|
|
|
- const channel_id = ChannelId.make(thread.parentId ?? thread.id)
|
|
|
+ const thread = asThreadChannel(message.channel)
|
|
|
+ if (!thread) return
|
|
|
+ const threadId = ThreadId.make(thread.id)
|
|
|
+ const channelId = ChannelId.make(thread.parentId ?? thread.id)
|
|
|
const allowed = isChannelAllowed(thread.parentId ?? "", thread.parent?.parentId ?? null, config)
|
|
|
|
|
|
if (!allowed) {
|
|
|
- const owned = yield* pool.hasTrackedThread(thread_id).pipe(
|
|
|
+ const owned = yield* sessions.hasTrackedThread(threadId).pipe(
|
|
|
Effect.catchAll(() => Effect.succeed(false)),
|
|
|
)
|
|
|
if (!owned || mentioned) {
|
|
|
- yield* ledger.setOffset(source, message.id)
|
|
|
+ yield* offsets.setOffset(source, message.id)
|
|
|
return
|
|
|
}
|
|
|
}
|
|
|
|
|
|
const event = ThreadMessage.make({
|
|
|
kind: "thread_message",
|
|
|
- thread_id,
|
|
|
- channel_id,
|
|
|
- message_id: message.id,
|
|
|
- guild_id: GuildId.make(message.guildId ?? ""),
|
|
|
- bot_user_id,
|
|
|
- bot_role_id,
|
|
|
- author_id: message.author.id,
|
|
|
- author_is_bot: message.author.bot,
|
|
|
- mentions_everyone: message.mentions.everyone,
|
|
|
+ threadId,
|
|
|
+ channelId,
|
|
|
+ messageId: message.id,
|
|
|
+ guildId: GuildId.make(message.guildId ?? ""),
|
|
|
+ botUserId,
|
|
|
+ botRoleId,
|
|
|
+ authorId: message.author.id,
|
|
|
+ authorIsBot: message.author.bot,
|
|
|
+ mentionsEveryone: message.mentions.everyone,
|
|
|
mentions,
|
|
|
content,
|
|
|
})
|
|
|
yield* offer(
|
|
|
event,
|
|
|
Effect.sync(() => {
|
|
|
- chats.set(event.thread_id, thread)
|
|
|
- stash(refs, ref_ids, event.message_id, message)
|
|
|
+ chats.set(event.threadId, thread)
|
|
|
+ stash(refs, ref_ids, event.messageId, message)
|
|
|
}),
|
|
|
)
|
|
|
- yield* ledger.setOffset(source, message.id)
|
|
|
+ yield* offsets.setOffset(source, message.id)
|
|
|
return
|
|
|
}
|
|
|
|
|
|
- const channel = message.channel as TextChannel
|
|
|
+ const channel = asTextChannel(message.channel)
|
|
|
+ if (!channel) return
|
|
|
if (!isChannelAllowed(channel.id, channel.parentId ?? null, config)) {
|
|
|
- yield* ledger.setOffset(source, message.id)
|
|
|
+ yield* offsets.setOffset(source, message.id)
|
|
|
return
|
|
|
}
|
|
|
|
|
|
const event = ChannelMessage.make({
|
|
|
kind: "channel_message",
|
|
|
- channel_id: ChannelId.make(channel.id),
|
|
|
- message_id: message.id,
|
|
|
- guild_id: GuildId.make(message.guildId ?? ""),
|
|
|
- bot_user_id,
|
|
|
- bot_role_id,
|
|
|
- author_id: message.author.id,
|
|
|
- author_is_bot: message.author.bot,
|
|
|
- mentions_everyone: message.mentions.everyone,
|
|
|
+ channelId: ChannelId.make(channel.id),
|
|
|
+ messageId: message.id,
|
|
|
+ guildId: GuildId.make(message.guildId ?? ""),
|
|
|
+ botUserId,
|
|
|
+ botRoleId,
|
|
|
+ authorId: message.author.id,
|
|
|
+ authorIsBot: message.author.bot,
|
|
|
+ mentionsEveryone: message.mentions.everyone,
|
|
|
mentions,
|
|
|
content,
|
|
|
})
|
|
|
yield* offer(
|
|
|
event,
|
|
|
Effect.sync(() => {
|
|
|
- texts.set(event.channel_id, channel)
|
|
|
- stash(refs, ref_ids, event.message_id, message)
|
|
|
+ texts.set(event.channelId, channel)
|
|
|
+ stash(refs, ref_ids, event.messageId, message)
|
|
|
}),
|
|
|
)
|
|
|
- yield* ledger.setOffset(source, message.id)
|
|
|
+ yield* offsets.setOffset(source, message.id)
|
|
|
})
|
|
|
|
|
|
const onMessage = (message: Message): void => {
|
|
|
- if (!mark(message.id)) return
|
|
|
const run = ingestMessage(message).pipe(
|
|
|
Effect.catchAll((error) =>
|
|
|
Effect.logError("Failed ingesting Discord message").pipe(
|
|
|
@@ -347,38 +207,28 @@ export class DiscordConversationServices {
|
|
|
void Runtime.runPromise(runtime)(run)
|
|
|
}
|
|
|
|
|
|
- const pullAfter = (channel: ChatChannel, after: string): Effect.Effect<number, unknown> =>
|
|
|
- Effect.tryPromise(() =>
|
|
|
- channel.messages.fetch({
|
|
|
- limit: CATCHUP_PAGE_SIZE,
|
|
|
- after,
|
|
|
- })
|
|
|
- ).pipe(
|
|
|
- Effect.map((page) => [...page.values()].sort((a, b) => a.createdTimestamp - b.createdTimestamp)),
|
|
|
- Effect.flatMap((rows) => {
|
|
|
- if (rows.length === 0) return Effect.succeed(0)
|
|
|
- const last = rows.at(-1)
|
|
|
- if (!last) return Effect.succeed(0)
|
|
|
- return Effect.forEach(rows, (row) => ingestMessage(row), { discard: true }).pipe(
|
|
|
- Effect.zipRight(
|
|
|
- rows.length < CATCHUP_PAGE_SIZE
|
|
|
- ? Effect.succeed(rows.length)
|
|
|
- : pullAfter(channel, last.id).pipe(Effect.map((tail: number) => rows.length + tail)),
|
|
|
- ),
|
|
|
- )
|
|
|
- }),
|
|
|
- )
|
|
|
-
|
|
|
const catchupSource = (source: string, channel: ChatChannel) =>
|
|
|
- Effect.gen(function* () {
|
|
|
- const offset = yield* ledger.getOffset(source)
|
|
|
- if (Option.isNone(offset)) {
|
|
|
- const page = yield* Effect.tryPromise(() => channel.messages.fetch({ limit: 1 }))
|
|
|
- const latest = page.first()
|
|
|
- if (latest) yield* ledger.setOffset(source, latest.id)
|
|
|
- return 0
|
|
|
- }
|
|
|
- return yield* pullAfter(channel, offset.value)
|
|
|
+ catchupFromOffset({
|
|
|
+ source,
|
|
|
+ pageSize: CATCHUP_PAGE_SIZE,
|
|
|
+ offsets,
|
|
|
+ fetchLatest: Effect.tryPromise(() => channel.messages.fetch({ limit: 1 })).pipe(
|
|
|
+ Effect.map((page) => {
|
|
|
+ const latest = page.first()
|
|
|
+ return latest ? Option.some(latest) : Option.none()
|
|
|
+ }),
|
|
|
+ ),
|
|
|
+ fetchAfter: (after) =>
|
|
|
+ Effect.tryPromise(() =>
|
|
|
+ channel.messages.fetch({
|
|
|
+ limit: CATCHUP_PAGE_SIZE,
|
|
|
+ after,
|
|
|
+ })
|
|
|
+ ).pipe(
|
|
|
+ Effect.map((page) => [...page.values()].sort((a, b) => a.createdTimestamp - b.createdTimestamp)),
|
|
|
+ ),
|
|
|
+ idOf: (message) => message.id,
|
|
|
+ ingest: (message) => ingestMessage(message),
|
|
|
})
|
|
|
|
|
|
const categoryChannels = () =>
|
|
|
@@ -404,14 +254,14 @@ export class DiscordConversationServices {
|
|
|
return nested.flat()
|
|
|
})
|
|
|
|
|
|
- const fetchText = (channel_id: string) =>
|
|
|
- Effect.tryPromise(() => client.channels.fetch(channel_id)).pipe(
|
|
|
+ const fetchText = (channelId: string) =>
|
|
|
+ Effect.tryPromise(() => client.channels.fetch(channelId)).pipe(
|
|
|
Effect.map((channel) => asTextChannel(channel)),
|
|
|
Effect.catchAll(() => Effect.succeed(null)),
|
|
|
)
|
|
|
|
|
|
- const fetchThread = (thread_id: string) =>
|
|
|
- Effect.tryPromise(() => client.channels.fetch(thread_id)).pipe(
|
|
|
+ const fetchThread = (threadId: string) =>
|
|
|
+ Effect.tryPromise(() => client.channels.fetch(threadId)).pipe(
|
|
|
Effect.map((channel) => asThreadChannel(channel)),
|
|
|
Effect.catchAll(() => Effect.succeed(null)),
|
|
|
)
|
|
|
@@ -424,18 +274,18 @@ export class DiscordConversationServices {
|
|
|
|
|
|
const fromChannels = yield* Effect.forEach(
|
|
|
channels,
|
|
|
- (channel_id) =>
|
|
|
- fetchText(channel_id).pipe(
|
|
|
+ (channelId) =>
|
|
|
+ fetchText(channelId).pipe(
|
|
|
Effect.flatMap((channel) => {
|
|
|
if (!channel) return Effect.succeed(0)
|
|
|
- return catchupSource(sourceChannel(channel_id), channel)
|
|
|
+ return catchupSource(sourceChannel(channelId), channel)
|
|
|
}),
|
|
|
Effect.catchAll((error) => {
|
|
|
const log = catchupBenign(error) ? Effect.logDebug("Channel catch-up skipped") : Effect.logWarning("Channel catch-up failed")
|
|
|
return log.pipe(
|
|
|
Effect.annotateLogs({
|
|
|
event: "conversation.catchup.channel.failed",
|
|
|
- channel_id,
|
|
|
+ channel_id: channelId,
|
|
|
error: messageOf(error),
|
|
|
}),
|
|
|
Effect.as(0),
|
|
|
@@ -447,18 +297,18 @@ export class DiscordConversationServices {
|
|
|
|
|
|
const fromThreads = yield* Effect.forEach(
|
|
|
threads,
|
|
|
- (thread_id) =>
|
|
|
- fetchThread(thread_id).pipe(
|
|
|
+ (threadId) =>
|
|
|
+ fetchThread(threadId).pipe(
|
|
|
Effect.flatMap((thread) => {
|
|
|
if (!thread) return Effect.succeed(0)
|
|
|
- return catchupSource(sourceThread(thread_id), thread)
|
|
|
+ return catchupSource(sourceThread(threadId), thread)
|
|
|
}),
|
|
|
Effect.catchAll((error) => {
|
|
|
const log = catchupBenign(error) ? Effect.logDebug("Thread catch-up skipped") : Effect.logWarning("Thread catch-up failed")
|
|
|
return log.pipe(
|
|
|
Effect.annotateLogs({
|
|
|
event: "conversation.catchup.thread.failed",
|
|
|
- thread_id,
|
|
|
+ thread_id: threadId,
|
|
|
error: messageOf(error),
|
|
|
}),
|
|
|
Effect.as(0),
|
|
|
@@ -492,7 +342,6 @@ export class DiscordConversationServices {
|
|
|
if (!interaction.isChatInputCommand()) return
|
|
|
const text = commandText(interaction.commandName)
|
|
|
if (!text) return
|
|
|
- if (!mark(interaction.id)) return
|
|
|
const handle = Effect.gen(function* () {
|
|
|
yield* Effect.tryPromise(() =>
|
|
|
interaction.deferReply({
|
|
|
@@ -504,11 +353,11 @@ export class DiscordConversationServices {
|
|
|
yield* acknowledge(interaction, COMMAND_NOT_THREAD_REPLY)
|
|
|
return
|
|
|
}
|
|
|
- const thread_id = ThreadId.make(thread.id)
|
|
|
- const channel_id = ChannelId.make(thread.parentId ?? thread.id)
|
|
|
+ const threadId = ThreadId.make(thread.id)
|
|
|
+ const channelId = ChannelId.make(thread.parentId ?? thread.id)
|
|
|
const allowed = isChannelAllowed(thread.parentId ?? "", thread.parent?.parentId ?? null, config)
|
|
|
if (!allowed) {
|
|
|
- const owned = yield* pool.hasTrackedThread(thread_id).pipe(
|
|
|
+ const owned = yield* sessions.hasTrackedThread(threadId).pipe(
|
|
|
Effect.catchAll(() => Effect.succeed(false)),
|
|
|
)
|
|
|
if (!owned) {
|
|
|
@@ -525,35 +374,28 @@ export class DiscordConversationServices {
|
|
|
return
|
|
|
}
|
|
|
|
|
|
- const bot_user_id = client.user?.id ?? ""
|
|
|
+ const botUserId = client.user?.id ?? ""
|
|
|
const event = ThreadMessage.make({
|
|
|
kind: "thread_message",
|
|
|
- thread_id,
|
|
|
- channel_id,
|
|
|
- message_id: interaction.id,
|
|
|
- guild_id: GuildId.make(interaction.guildId ?? ""),
|
|
|
- bot_user_id,
|
|
|
- bot_role_id: config.discordRoleId,
|
|
|
- author_id: interaction.user.id,
|
|
|
- author_is_bot: false,
|
|
|
- mentions_everyone: false,
|
|
|
+ threadId,
|
|
|
+ channelId,
|
|
|
+ messageId: interaction.id,
|
|
|
+ guildId: GuildId.make(interaction.guildId ?? ""),
|
|
|
+ botUserId,
|
|
|
+ botRoleId: config.discordRoleId,
|
|
|
+ authorId: interaction.user.id,
|
|
|
+ authorIsBot: false,
|
|
|
+ mentionsEveryone: false,
|
|
|
mentions: Mention.make({
|
|
|
- user_ids: bot_user_id.length > 0 ? [bot_user_id] : [],
|
|
|
- role_ids: [],
|
|
|
+ userIds: botUserId.length > 0 ? [botUserId] : [],
|
|
|
+ roleIds: [],
|
|
|
}),
|
|
|
content: text,
|
|
|
})
|
|
|
- const ingest = ledger.dedup(event.message_id).pipe(
|
|
|
- Effect.flatMap((fresh) => {
|
|
|
- if (!fresh) return Effect.void
|
|
|
- return Effect.sync(() => {
|
|
|
- chats.set(event.thread_id, thread)
|
|
|
- input.unsafeOffer(event)
|
|
|
- })
|
|
|
- }),
|
|
|
- Effect.catchAll(() => Effect.void),
|
|
|
- )
|
|
|
- yield* ingest
|
|
|
+ yield* Effect.sync(() => {
|
|
|
+ chats.set(event.threadId, thread)
|
|
|
+ input.unsafeOffer(event)
|
|
|
+ })
|
|
|
yield* acknowledge(interaction, COMMAND_ACK)
|
|
|
})
|
|
|
void Runtime.runPromise(runtime)(handle)
|
|
|
@@ -621,18 +463,18 @@ export class DiscordConversationServices {
|
|
|
events: Stream.fromQueue(input, { shutdown: false }),
|
|
|
})
|
|
|
|
|
|
- const channelOf = (thread_id: ThreadId, action: Action["kind"]) => {
|
|
|
- const channel = chats.get(thread_id)
|
|
|
+ const channelOf = (threadId: ThreadId, action: Action["kind"]) => {
|
|
|
+ const channel = chats.get(threadId)
|
|
|
if (channel) return Effect.succeed(channel)
|
|
|
- return Effect.tryPromise(() => client.channels.fetch(thread_id)).pipe(
|
|
|
+ return Effect.tryPromise(() => client.channels.fetch(threadId)).pipe(
|
|
|
Effect.flatMap((fetched) => {
|
|
|
const thread = asThreadChannel(fetched)
|
|
|
if (thread) {
|
|
|
- chats.set(thread_id, thread)
|
|
|
+ chats.set(threadId, thread)
|
|
|
return Effect.succeed(thread)
|
|
|
}
|
|
|
return DeliveryError.make({
|
|
|
- thread_id,
|
|
|
+ threadId,
|
|
|
action,
|
|
|
message: "missing-thread-channel",
|
|
|
retriable: false,
|
|
|
@@ -640,7 +482,7 @@ export class DiscordConversationServices {
|
|
|
}),
|
|
|
Effect.mapError((cause) =>
|
|
|
DeliveryError.make({
|
|
|
- thread_id,
|
|
|
+ threadId,
|
|
|
action,
|
|
|
message: messageOf(cause),
|
|
|
retriable: deliveryRetriable(cause),
|
|
|
@@ -648,13 +490,13 @@ export class DiscordConversationServices {
|
|
|
)
|
|
|
}
|
|
|
|
|
|
- const deliver = (thread_id: ThreadId, action: Action["kind"], send: Effect.Effect<unknown, unknown>) =>
|
|
|
+ const deliver = (threadId: ThreadId, action: Action["kind"], send: Effect.Effect<unknown, unknown>) =>
|
|
|
Effect.gen(function* () {
|
|
|
const attempts = yield* Ref.make(0)
|
|
|
yield* send.pipe(
|
|
|
Effect.mapError((cause) =>
|
|
|
DeliveryError.make({
|
|
|
- thread_id,
|
|
|
+ threadId,
|
|
|
action,
|
|
|
message: messageOf(cause),
|
|
|
retriable: deliveryRetriable(cause),
|
|
|
@@ -665,7 +507,7 @@ export class DiscordConversationServices {
|
|
|
Effect.logWarning("Discord delivery attempt failed").pipe(
|
|
|
Effect.annotateLogs({
|
|
|
event: "conversation.delivery.retry",
|
|
|
- thread_id,
|
|
|
+ thread_id: threadId,
|
|
|
action,
|
|
|
attempt,
|
|
|
retriable: error.retriable,
|
|
|
@@ -681,7 +523,7 @@ export class DiscordConversationServices {
|
|
|
Effect.logError("Discord delivery failed").pipe(
|
|
|
Effect.annotateLogs({
|
|
|
event: "conversation.delivery.failed",
|
|
|
- thread_id,
|
|
|
+ thread_id: threadId,
|
|
|
action,
|
|
|
attempts: attempt,
|
|
|
retriable: error.retriable,
|
|
|
@@ -693,34 +535,34 @@ export class DiscordConversationServices {
|
|
|
)
|
|
|
})
|
|
|
|
|
|
- const sendTyping = (thread_id: ThreadId) =>
|
|
|
+ const sendTyping = (threadId: ThreadId) =>
|
|
|
Effect.gen(function* () {
|
|
|
- const channel = yield* channelOf(thread_id, "typing")
|
|
|
- yield* deliver(thread_id, "typing", Effect.tryPromise(() => channel.sendTyping()))
|
|
|
+ const channel = yield* channelOf(threadId, "typing")
|
|
|
+ yield* deliver(threadId, "typing", Effect.tryPromise(() => channel.sendTyping()))
|
|
|
})
|
|
|
|
|
|
- const sendText = (thread_id: ThreadId, action: "send" | "reply", text: string) =>
|
|
|
+ const sendText = (threadId: ThreadId, action: "send" | "reply", text: string) =>
|
|
|
Effect.gen(function* () {
|
|
|
- const channel = yield* channelOf(thread_id, action)
|
|
|
+ const channel = yield* channelOf(threadId, action)
|
|
|
yield* Effect.forEach(
|
|
|
splitForDiscord(cleanResponse(text)),
|
|
|
- (chunk) => deliver(thread_id, action, Effect.tryPromise(() => channel.send(chunk))),
|
|
|
+ (chunk) => deliver(threadId, action, Effect.tryPromise(() => channel.send(chunk))),
|
|
|
{ discard: true },
|
|
|
)
|
|
|
})
|
|
|
|
|
|
const publish = (action: Action) => {
|
|
|
- if (action.kind === "typing") return sendTyping(action.thread_id)
|
|
|
- return sendText(action.thread_id, action.kind, action.text)
|
|
|
+ if (action.kind === "typing") return sendTyping(action.threadId)
|
|
|
+ return sendText(action.threadId, action.kind, action.text)
|
|
|
}
|
|
|
|
|
|
- const withTyping = <A, E, R>(thread_id: ThreadId, self: Effect.Effect<A, E, R>) =>
|
|
|
+ const withTyping = <A, E, R>(threadId: ThreadId, self: Effect.Effect<A, E, R>) =>
|
|
|
Effect.scoped(
|
|
|
Effect.gen(function* () {
|
|
|
const pulse = publish(
|
|
|
Typing.make({
|
|
|
kind: "typing",
|
|
|
- thread_id,
|
|
|
+ threadId,
|
|
|
}),
|
|
|
).pipe(Effect.catchAll(() => Effect.void))
|
|
|
yield* pulse
|
|
|
@@ -736,14 +578,14 @@ export class DiscordConversationServices {
|
|
|
const outbox = Outbox.of({ publish, withTyping })
|
|
|
|
|
|
const history = History.of({
|
|
|
- rehydrate: (thread_id, latest: string) =>
|
|
|
+ rehydrate: (threadId, latest: string) =>
|
|
|
Effect.gen(function* () {
|
|
|
- const channel = chats.get(thread_id)
|
|
|
+ const channel = chats.get(threadId)
|
|
|
if (!channel) return latest
|
|
|
return yield* buildHistoryReplayPrompt(channel, latest).pipe(
|
|
|
Effect.mapError((cause) =>
|
|
|
HistoryError.make({
|
|
|
- thread_id,
|
|
|
+ threadId,
|
|
|
message: messageOf(cause),
|
|
|
retriable: true,
|
|
|
})),
|
|
|
@@ -754,40 +596,40 @@ export class DiscordConversationServices {
|
|
|
const threads = Threads.of({
|
|
|
ensure: (event, name: string) => {
|
|
|
if (event.kind === "thread_message") {
|
|
|
- return Effect.succeed(ThreadRef.make({ thread_id: event.thread_id, channel_id: event.channel_id }))
|
|
|
+ return Effect.succeed(ThreadRef.make({ threadId: event.threadId, channelId: event.channelId }))
|
|
|
}
|
|
|
|
|
|
- const known = roots.get(event.message_id)
|
|
|
+ const known = roots.get(event.messageId)
|
|
|
if (known) {
|
|
|
- return Effect.succeed(ThreadRef.make({ thread_id: known, channel_id: event.channel_id }))
|
|
|
+ return Effect.succeed(ThreadRef.make({ threadId: known, channelId: event.channelId }))
|
|
|
}
|
|
|
|
|
|
return Effect.gen(function* () {
|
|
|
- const local = texts.get(event.channel_id)
|
|
|
+ const local = texts.get(event.channelId)
|
|
|
const channel = local
|
|
|
? local
|
|
|
- : yield* Effect.tryPromise(() => client.channels.fetch(event.channel_id)).pipe(
|
|
|
+ : yield* Effect.tryPromise(() => client.channels.fetch(event.channelId)).pipe(
|
|
|
Effect.map((fetched) => asTextChannel(fetched)),
|
|
|
Effect.mapError((cause) =>
|
|
|
ThreadEnsureError.make({
|
|
|
- channel_id: event.channel_id,
|
|
|
+ channelId: event.channelId,
|
|
|
message: messageOf(cause),
|
|
|
retriable: deliveryRetriable(cause),
|
|
|
})),
|
|
|
)
|
|
|
if (!channel) {
|
|
|
return yield* ThreadEnsureError.make({
|
|
|
- channel_id: event.channel_id,
|
|
|
+ channelId: event.channelId,
|
|
|
message: "missing-parent-channel",
|
|
|
retriable: false,
|
|
|
})
|
|
|
}
|
|
|
- texts.set(event.channel_id, channel)
|
|
|
- const base = refs.get(event.message_id)
|
|
|
+ texts.set(event.channelId, channel)
|
|
|
+ const base = refs.get(event.messageId)
|
|
|
const thread = yield* Effect.tryPromise(() =>
|
|
|
channel.threads.create({
|
|
|
name,
|
|
|
- startMessage: base ?? event.message_id,
|
|
|
+ startMessage: base ?? event.messageId,
|
|
|
autoArchiveDuration: 60,
|
|
|
}),
|
|
|
).pipe(
|
|
|
@@ -801,15 +643,15 @@ export class DiscordConversationServices {
|
|
|
)),
|
|
|
Effect.mapError((cause) =>
|
|
|
ThreadEnsureError.make({
|
|
|
- channel_id: event.channel_id,
|
|
|
+ channelId: event.channelId,
|
|
|
message: messageOf(cause),
|
|
|
retriable: deliveryRetriable(cause),
|
|
|
})),
|
|
|
)
|
|
|
- const thread_id = ThreadId.make(thread.id)
|
|
|
- chats.set(thread_id, thread)
|
|
|
- stash(roots, root_ids, event.message_id, thread_id)
|
|
|
- return ThreadRef.make({ thread_id, channel_id: event.channel_id })
|
|
|
+ const threadId = ThreadId.make(thread.id)
|
|
|
+ chats.set(threadId, thread)
|
|
|
+ stash(roots, root_ids, event.messageId, threadId)
|
|
|
+ return ThreadRef.make({ threadId, channelId: event.channelId })
|
|
|
})
|
|
|
},
|
|
|
})
|