Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 20 additions & 15 deletions apps/cluster/src/Chat/Discord/Api.ts
Original file line number Diff line number Diff line change
Expand Up @@ -115,15 +115,29 @@ export interface DiscordMember {
readonly roles?: ReadonlyArray<string> | undefined
}

/**
* The author is admitted with the role in `member.roles`; a bot account,
* the janitor bot included, is context only whatever its roles. The name is
* the server nickname, else the display name, else the username.
*/
export const toAuthor = (
user: Pick<Discord.UserResponse, "id" | "username" | "global_name" | "bot">,
member: DiscordMember | undefined,
roleId: string,
): ChatMessage["author"] => ({
id: user.id,
name: member?.nick ?? user.global_name ?? user.username,
class:
user.bot === true ? "bot" : member?.roles?.includes(roleId) === true ? "admitted" : "outsider",
})

/** Mentions of the bot, `<@id>` or the older `<@!id>`. */
const botMention = (botUserId: string) => new RegExp(`<@!?${botUserId}>`, "g")

/**
* The author is admitted with the role in `member.roles`; a bot account,
* the janitor bot included, is context only whatever its roles. The name is
* the server nickname, else the display name, else the username. A reply to
* one of the bot's messages counts as a mention: it is addressed to the
* bot whether or not the reply pings (EFF-3137).
* The author as `toAuthor` classes them. A reply to one of the bot's
* messages counts as a mention: it is addressed to the bot whether or not
* the reply pings (EFF-3137).
*/
export const toChatMessage = (
message: DiscordMessage,
Expand All @@ -136,16 +150,7 @@ export const toChatMessage = (
const answers = message.referenced_message?.author.id
return {
id: message.id,
author: {
id: message.author.id,
name: member?.nick ?? message.author.global_name ?? message.author.username,
class:
message.author.bot === true
? "bot"
: member?.roles?.includes(roleId) === true
? "admitted"
: "outsider",
},
author: toAuthor(message.author, member, roleId),
text: (mentionsBotRole ? message.content.replaceAll(`<@&${botRoleId}>`, "") : message.content)
.replace(botMention(botUserId), "")
.trim(),
Expand Down
24 changes: 15 additions & 9 deletions apps/cluster/src/Chat/Discord/Gateway.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,23 +30,25 @@ import * as Ingress from "./Ingress.ts"
* wakes the object after an eviction and restarts a shard that died or went
* quiet.
*
* It is ingress only: each `MESSAGE_CREATE` goes through `Ingress` before
* its sequence is stored, so a message that couldn't reach its session comes
* back with the RESUME, and the message ids already handled are kept here to
* drop the ones a RESUME replays. A 12-hour spike measured two evictions with
* a live socket in six hours, each healed by the alarm within a minute, and
* every reconnect was a RESUME.
* It is ingress only: each `MESSAGE_CREATE` and `MESSAGE_REACTION_ADD` goes
* through `Ingress` before its sequence is stored, so a message that
* couldn't reach its session comes back with the RESUME, and the message
* ids already handled are kept here to drop the ones a RESUME replays. A
* 12-hour spike measured two evictions with a live socket in six hours,
* each healed by the alarm within a minute, and every reconnect was a
* RESUME.
*/

export const GATEWAY_NAME = "gateway"

/**
* `GUILDS`, `GUILD_MESSAGES`, `DIRECT_MESSAGES` (for the pointer) and the
* privileged `MESSAGE_CONTENT`.
* `GUILDS`, `GUILD_MESSAGES`, `GUILD_MESSAGE_REACTIONS` (for approvals),
* `DIRECT_MESSAGES` (for the pointer) and the privileged `MESSAGE_CONTENT`.
*/
export const INTENTS =
GatewayIntentBits.Guilds |
GatewayIntentBits.GuildMessages |
GatewayIntentBits.GuildMessageReactions |
GatewayIntentBits.DirectMessages |
GatewayIntentBits.MessageContent

Expand Down Expand Up @@ -382,7 +384,11 @@ export const ChatDiscordGatewayLive = (sessions: Layer.Layer<Ingress.DiscordSess
]),
// Runs before the sequence is stored: a failure reconnects and resumes from it.
Effect.provideService(DispatchHook, (payload) =>
payload.t === "MESSAGE_CREATE" ? Effect.orDie(ingress.handle(payload.d)) : Effect.void,
payload.t === "MESSAGE_CREATE"
? Effect.orDie(ingress.handle(payload.d))
: payload.t === "MESSAGE_REACTION_ADD"
? Effect.orDie(ingress.handleReaction(payload.d))
: Effect.void,
),
Effect.provideService(References.MinimumLogLevel, "Trace"),
Effect.provideService(References.CurrentLoggers, new Set([watcher])),
Expand Down
86 changes: 76 additions & 10 deletions apps/cluster/src/Chat/Discord/Ingress.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,11 @@ import { DiscordREST, type DiscordRESTError } from "dfx/DiscordREST"
import type * as Discord from "dfx/types"
import type * as Sharding from "effect/cluster/Sharding"
import * as Context from "effect/Context"
import * as DateTime from "effect/DateTime"
import * as Effect from "effect/Effect"
import * as Layer from "effect/Layer"
import * as Schema from "effect/Schema"
import type { ChatMessage, ChatThread } from "../Platform.ts"
import { APPROVAL_TEXT, approvalMessageId, type ChatMessage, type ChatThread } from "../Platform.ts"
import { sessionClient } from "../Session.ts"
import {
errorCode,
Expand All @@ -15,6 +16,7 @@ import {
SUPPRESS_EMBEDS,
THREAD_PARENT_TYPES,
THREAD_TYPES,
toAuthor,
toChatMessage,
} from "./Api.ts"
import { DiscordBotConfig } from "./Config.ts"
Expand Down Expand Up @@ -54,10 +56,10 @@ export class DiscordSessions extends Context.Service<
}

/**
* The message ids the gateway has handed on, kept in the gateway Durable
* Object's storage. A RESUME replays what arrived while the socket was dark,
* and a rebuilt instance reads the same storage, so a replayed message is
* dropped.
* The message ids the gateway has handed on, approvals' included, kept in
* the gateway Durable Object's storage. A RESUME replays what arrived while
* the socket was dark, and a rebuilt instance reads the same storage, so a
* replayed message is dropped.
*/
export class SeenMessages extends Context.Service<
SeenMessages,
Expand All @@ -77,20 +79,31 @@ export class DiscordIngressError extends Schema.TaggedError<DiscordIngressError>
/**
* The gateway Durable Object's ingress: it filters a `MESSAGE_CREATE`,
* admits its author by `member.roles` and hands the message to its thread's
* session, creating the thread for a channel mention.
* session, creating the thread for a channel mention. A `MESSAGE_REACTION_ADD`
* of 👍 on the bot's message becomes an approval (EFF-3163).
*/
export class DiscordIngress extends Context.Service<
DiscordIngress,
{
readonly handle: (
message: Discord.GatewayMessageCreateDispatchData,
) => Effect.Effect<void, DiscordIngressError>
readonly handleReaction: (
reaction: Discord.GatewayMessageReactionAddDispatchData,
) => Effect.Effect<void, DiscordIngressError>
}
>()("@janitor/cluster/Chat/Discord/Ingress/DiscordIngress") {}

/** The message ingress logs at Debug for each message it drops, with the reason. */
export const DROPPED = "Discord ingress dropped a message"

/**
* 👍 with any skin tone. Discord sends a Unicode emoji with a null id,
* though dfx types a reaction's emoji as a custom one, which has an id.
*/
const isThumbsUp = (emoji: { readonly id: string | null; readonly name: string | null }) =>
emoji.id === null && emoji.name?.startsWith("👍") === true

const DIRECT_MESSAGE_POINTER = "I only work in servers. Mention me in a channel to start a thread."

/** Threads keep a day of quiet before Discord archives them. */
Expand Down Expand Up @@ -202,22 +215,73 @@ const make = Effect.gen(function* () {
)
})

/**
* Hands an admitted member's 👍 on the bot's message in a thread with a
* session to it as an approval, or drops it. A reaction never starts a
* session, and anyone else's is dropped, not kept as context. Only an
* approval handed on is recorded as seen: a dropped reaction, an
* outsider's say, mustn't suppress a later 👍 from the same user.
*/
const routeReaction = Effect.fnUntraced(function* (
reaction: Discord.GatewayMessageReactionAddDispatchData,
id: string,
) {
const bot = yield* botUserId
if (reaction.message_author_id !== bot) return yield* drop("a reaction to another's message")
if (reaction.guild_id !== config.guildId) return yield* drop("a reaction outside the guild")
if (reaction.member === undefined) return yield* drop("no member")
const author = toAuthor(reaction.member.user, reaction.member, config.roleId)
if (author.class !== "admitted") return yield* drop(`a reaction from a ${author.class}`)
if (!THREAD_TYPES.has(yield* channelType(reaction.channel_id)))
return yield* drop("a reaction outside a thread")
const thread: ChatThread = {
_tag: "Discord",
guild: reaction.guild_id,
threadId: reaction.channel_id,
}
if (!(yield* sessions.exists(thread)))
return yield* drop("a reaction in a thread with no session")
yield* sessions.receive(thread, {
id,
author,
text: APPROVAL_TEXT,
mentionsBot: true,
sentAt: yield* DateTime.now,
attachments: [],
})
yield* seen.add(id)
yield* Effect.logInfo("Discord ingress handed an approval to its session").pipe(
Effect.annotateLogs({ threadId: thread.threadId }),
)
})

/** A refusal from Discord won't change on a replay: it's logged and the message dropped. */
const refused = (error: DiscordRESTError) =>
Effect.logWarning("Discord refused a request for an inbound message").pipe(
Effect.annotateLogs({ reason: error.message }),
)

/** Routes, dropping what Discord refuses; any other failure is for the replay. */
const attempt = (routed: Effect.Effect<void, DiscordRESTError | DiscordSessionError>) =>
routed.pipe(
Effect.catchTags({ ErrorResponse: refused, RatelimitedResponse: refused }),
Effect.mapError((error) => new DiscordIngressError({ message: error.message })),
)

return DiscordIngress.of({
handle: (message) =>
Effect.gen(function* () {
if (yield* seen.has(message.id)) return yield* drop("already handled")
yield* route(message).pipe(
Effect.catchTags({ ErrorResponse: refused, RatelimitedResponse: refused }),
Effect.mapError((error) => new DiscordIngressError({ message: error.message })),
)
yield* attempt(route(message))
yield* seen.add(message.id)
}).pipe(Effect.annotateLogs({ discordMessageId: message.id })),
handleReaction: (reaction) =>
Effect.gen(function* () {
if (!isThumbsUp(reaction.emoji)) return yield* drop("a reaction other than a thumbs up")
const id = approvalMessageId(reaction.message_id, reaction.user_id)
if (yield* seen.has(id)) return yield* drop("already handled")
yield* attempt(routeReaction(reaction, id))
}).pipe(Effect.annotateLogs({ discordMessageId: reaction.message_id })),
})
})

Expand All @@ -227,6 +291,8 @@ const make = Effect.gen(function* () {
* mention starts a thread from the message and hands it to that thread's
* session; in a thread, a message reaches the session when it has one or
* when an admitted user mentions the bot. A direct message gets a pointer.
* An admitted user's 👍 on the bot's message in a thread with a session is
* handed to it as an approval that counts as a mention.
*/
export const layer: Layer.Layer<
DiscordIngress,
Expand Down
14 changes: 11 additions & 3 deletions apps/cluster/src/Chat/Discord/Platform.ts
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,9 @@ const UNKNOWN_MEMBER = 10007
/** Discord's code for "Unknown Message". */
const UNKNOWN_MESSAGE = 10008

/** A Discord id. */
const SNOWFLAKE = /^\d+$/

/** The message a thread started from a message opens with, pointing at it. */
const THREAD_STARTER_MESSAGE = 21

Expand Down Expand Up @@ -191,10 +194,15 @@ const make = Effect.gen(function* () {
const replyTo = options?.replyTo
// A thread started from a message shares its id, and that message
// lives in the parent channel, where a reply from inside the thread
// can't point: it would show "Message could not be loaded". Its
// author is mentioned instead, so they are still pinged.
// can't point: it would show "Message could not be loaded". An
// approval's id (EFF-3163) is no message's, and Discord refuses it.
// The author is mentioned instead, so they are still pinged.
const starter =
thread._tag === "Discord" && replyTo?.messageId === thread.threadId ? replyTo : undefined
thread._tag === "Discord" &&
replyTo !== undefined &&
(replyTo.messageId === thread.threadId || !SNOWFLAKE.test(replyTo.messageId))
? replyTo
: undefined
const reference =
replyTo === undefined || starter !== undefined
? undefined
Expand Down
11 changes: 11 additions & 0 deletions apps/cluster/src/Chat/Platform.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,17 @@ export const sessionKey = (thread: ChatThread): string =>
? `slack/${thread.team}/${thread.channel}/${thread.threadTs}`
: `discord/${thread.guild}/${thread.threadId}`

/**
* What an admitted user's 👍 on one of the bot's messages hands the session
* (EFF-3163): a message that counts as a mention, so a yes/no question or a
* request for approval can be answered with a reaction.
*/
export const APPROVAL_TEXT = "👍 (reacted with a thumbs up to your message: approved)"

/** An approval's message id, unlike any platform message id, so sessions store each once. */
export const approvalMessageId = (messageId: string, userId: string): string =>
`reaction/${messageId}/${userId}`

export class ChatPlatformError extends Schema.TaggedError<ChatPlatformError>()(
"ChatPlatformError",
{ message: Schema.String },
Expand Down
9 changes: 7 additions & 2 deletions apps/cluster/src/Chat/Slack/Api.ts
Original file line number Diff line number Diff line change
Expand Up @@ -226,7 +226,12 @@ export const SlackFile = Schema.Struct({
size: Schema.optionalKey(Schema.Finite),
})

const Timestamp = Schema.String.check(Schema.isPattern(/^\d+\.\d+$/))
/** A Slack `ts`: a message's id and the time it was sent. */
export const Timestamp = Schema.String.check(Schema.isPattern(/^\d+\.\d+$/))

/** When a Slack `ts` was sent. */
export const sentAt = (ts: string): DateTime.Utc =>
DateTime.makeUnsafe(Math.floor(Number(ts) * 1000))

/** The fields of a Slack message that janitor reads, in an event or a thread's history. */
export const SlackMessage = Schema.Struct({
Expand Down Expand Up @@ -266,7 +271,7 @@ export const toChatMessage = (
author: { id: author.id, name: author.name, class: author.class },
text: (message.text ?? "").replace(mentionPattern(botUserId), "").replace(/^\s+/, ""),
mentionsBot: mentioned,
sentAt: DateTime.makeUnsafe(Math.floor(Number(message.ts) * 1000)),
sentAt: sentAt(message.ts),
attachments: (message.files ?? []).map((file) => ({
id: file.id,
name: file.name ?? file.id,
Expand Down
Loading