Pin messages (GRYT-1619)

src/db/interfaces.ts

@@ -156,6 +156,9 @@ export interface MessageRecord {
   text_fallback?: boolean;
   /** What a webhook posted under. Enrichment turns it into sender_nickname. */
   sender_display_name?: string | null;
+  /** When it was pinned and by whom (GRYT-1619). Absent on a message that isn't. */
+  pinned_at?: Date | null;
+  pinned_by?: string | null;
   /** Set on the notice the server writes for apps from before MLS, sent as "system".
       Apps that read MLS hide these rows (GRYT-1508). */
   mls_placeholder?: MlsPlaceholder;

src/db/sqlite/connection.ts

@@ -1021,6 +1021,12 @@ function runMigrations(d: DatabaseSync): void {
     d.exec("ALTER TABLE messages ADD COLUMN mls_seq INTEGER");
     d.exec("ALTER TABLE messages ADD COLUMN mls_sender_server_id TEXT");
   }
+  // Pinned messages (GRYT-1619): when and by whom. NULL on a message that isn't pinned.
+  if (!hasColumn(d, "messages", "pinned_at")) {
+    d.exec("ALTER TABLE messages ADD COLUMN pinned_at TEXT");
+    d.exec("ALTER TABLE messages ADD COLUMN pinned_by TEXT");
+  }
+  d.exec("CREATE INDEX IF NOT EXISTS idx_messages_pinned ON messages(conversation_id, pinned_at) WHERE pinned_at IS NOT NULL");
   // One stored file per picture a webhook sends, however often it sends it.
   d.exec(`CREATE TABLE IF NOT EXISTS webhook_media (
     webhook_id TEXT NOT NULL,

src/db/sqlite/messages.ts

@@ -21,6 +21,7 @@ function rowToMessage(r: Record<string, unknown>): MessageRecord {
     ...(r.text_fallback ? { text_fallback: true } : {}),
     ...(r.sender_display_name ? { sender_display_name: r.sender_display_name as string } : {}),
     ...(r.sender_avatar_file_id ? { sender_avatar_file_id: r.sender_avatar_file_id as string } : {}),
+    ...(r.pinned_at ? { pinned_at: fromIso(r.pinned_at as string), pinned_by: (r.pinned_by as string) ?? null } : {}),
     ...(r.mls_seq != null
       ? { mls_placeholder: { seq: Number(r.mls_seq), sender_server_id: r.mls_sender_server_id as string } }
       : {}),
@@ -116,6 +117,32 @@ export async function updateMessageText(conversationId: string, messageId: strin
   return getMessageById(conversationId, messageId);
 }
 
+/** Pins a message, or unpins it with `by` null. Returns the message as it now is, or null if it's gone. */
+export async function setMessagePinned(conversationId: string, messageId: string, by: string | null): Promise<MessageRecord | null> {
+  const db = getSqliteDb();
+  const result = db
+    .prepare(`UPDATE messages SET pinned_at = ?, pinned_by = ? WHERE conversation_id = ? AND message_id = ?`)
+    .run(by ? toIso(new Date()) : null, by, conversationId, messageId);
+  if (result.changes === 0) return null;
+  return getMessageById(conversationId, messageId);
+}
+
+/** A conversation's pinned messages, newest pin first. */
+export async function listPinnedMessages(conversationId: string, limit = 50): Promise<MessageRecord[]> {
+  const db = getSqliteDb();
+  const rows = db
+    .prepare(`SELECT * FROM messages WHERE conversation_id = ? AND pinned_at IS NOT NULL ORDER BY pinned_at DESC LIMIT ?`)
+    .all(conversationId, limit) as Record<string, unknown>[];
+  return rows.map(rowToMessage);
+}
+
+/** How many a conversation has pinned, for the cap. */
+export async function countPinnedMessages(conversationId: string): Promise<number> {
+  const db = getSqliteDb();
+  const row = db.prepare(`SELECT COUNT(*) AS n FROM messages WHERE conversation_id = ? AND pinned_at IS NOT NULL`).get(conversationId) as { n: number };
+  return row.n;
+}
+
 export async function insertFile(
   // dominant_color is written later by the image worker, so callers inserting
   // a fresh upload do not supply one.

src/db/sqlite/pins.test.ts

@@ -0,0 +1,48 @@
+import assert from "node:assert/strict";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { after, before, describe, it } from "node:test";
+
+import { initSqlite } from "./connection";
+import { countPinnedMessages, getMessageById, insertMessage, listPinnedMessages, setMessagePinned } from "./messages";
+
+describe("pinned messages", () => {
+  const dir = mkdtempSync(join(tmpdir(), "gryt-pins-"));
+  const CONV = "c1";
+
+  before(async () => {
+    process.env.DATA_DIR = dir;
+    await initSqlite();
+    const base = Date.UTC(2026, 0, 1, 12, 0, 0);
+    for (const id of ["a", "b", "c"]) {
+      await insertMessage({ conversation_id: CONV, sender_server_id: "u1", text: id, attachments: null, reactions: null, message_id: id, created_at: new Date(base) });
+    }
+    await insertMessage({ conversation_id: "other", sender_server_id: "u1", text: "x", attachments: null, reactions: null, message_id: "x" });
+  });
+
+  after(() => {
+    rmSync(dir, { recursive: true, force: true });
+  });
+
+  it("pins, lists newest pin first, and unpins", async () => {
+    const a = await setMessagePinned(CONV, "a", "mod");
+    assert.equal(a?.pinned_by, "mod");
+    assert.ok(a?.pinned_at instanceof Date);
+    await new Promise((r) => setTimeout(r, 5));
+    await setMessagePinned(CONV, "c", "mod");
+    await setMessagePinned("other", "x", "mod");
+
+    assert.deepEqual((await listPinnedMessages(CONV)).map((m) => m.message_id), ["c", "a"]);
+    assert.equal(await countPinnedMessages(CONV), 2);
+
+    const un = await setMessagePinned(CONV, "a", null);
+    assert.equal(un?.pinned_at, undefined);
+    assert.equal((await getMessageById(CONV, "a"))?.pinned_by, undefined);
+    assert.deepEqual((await listPinnedMessages(CONV)).map((m) => m.message_id), ["c"]);
+  });
+
+  it("returns null for a message that isn't there", async () => {
+    assert.equal(await setMessagePinned(CONV, "missing", "mod"), null);
+  });
+});

src/socket/handlers/chat.ts

@@ -20,6 +20,9 @@ import {
   deleteMessage,
   getMessageById,
   updateMessageText,
+  setMessagePinned,
+  listPinnedMessages,
+  countPinnedMessages,
   getFilesByIds,
   getImageJobStatusForFile,
   getServerConfig,
@@ -86,6 +89,8 @@ import { DISCORD_SENDER_PREFIX } from "../../import/discord/ids";
 const RL_SEND: RateLimitRule = { limit: 20, windowMs: 10_000, banMs: 30_000, scorePerAction: 1, maxScore: 10, scoreDecayMs: 2000 };
 const RL_REACT: RateLimitRule = { limit: 60, windowMs: 60_000, scorePerAction: 0.5, maxScore: 15, scoreDecayMs: 3000 };
 const RL_DELETE: RateLimitRule = { limit: 30, windowMs: 60_000, scorePerAction: 1, maxScore: 15, scoreDecayMs: 3000 };
+/** Pins per conversation. Past this a pin list stops being the short list it's for. */
+const MAX_PINS = 50;
 const RL_EDIT: RateLimitRule = { limit: 20, windowMs: 60_000, scorePerAction: 1, maxScore: 10, scoreDecayMs: 2000 };
 const RL_FETCH: RateLimitRule = { limit: 15, windowMs: 10_000, scorePerAction: 0.3, maxScore: 8, scoreDecayMs: 1500 };
 
@@ -1498,5 +1503,93 @@ export function registerChatHandlers(ctx: HandlerContext): EventHandlerMap {
         socket.emit("chat:error", "Failed to edit message");
       }
     },
+
+    // Pin or unpin a message (GRYT-1619). In a channel that takes manage_messages, as it did on
+    // Discord; in a DM or group either side may, since there is nobody else to ask.
+    'chat:pin': async (payload: { conversationId: string; messageId: string; pinned: boolean; accessToken: string }) => {
+      try {
+        const ip = getClientIp();
+        const userId = clientsInfo[clientId]?.serverUserId;
+        const rl = checkRateLimit("chat:pin", userId, ip, RL_EDIT);
+        if (!rl.allowed) {
+          socket.emit("chat:error", { error: "rate_limited", retryAfterMs: rl.retryAfterMs, message: `Too fast. Wait ${Math.ceil((rl.retryAfterMs || 0) / 1000)}s.` });
+          return;
+        }
+        if (!payload || !payload.conversationId || !payload.messageId || typeof payload.pinned !== "boolean" || !payload.accessToken) {
+          socket.emit("chat:error", "Invalid pin payload");
+          return;
+        }
+
+        const auth = await requireAuth(socket, payload);
+        if (!auth) return;
+        const access = await requireConversationAccess(payload.conversationId, auth.tokenPayload.serverUserId);
+        if (!access) return;
+
+        if (access.kind !== "dm" && !(await mayHere(auth, payload.conversationId, "manage_messages"))) {
+          socket.emit("chat:error", {
+            error: "forbidden",
+            message: "You do not have permission to pin messages here.",
+            permission: "manage_messages",
+          });
+          return;
+        }
+
+        const message = await getMessageById(payload.conversationId, payload.messageId);
+        if (!message) { socket.emit("chat:error", "Message not found"); return; }
+        // Already as asked: answer with the state anyway, so a second click isn't an error.
+        const already = Boolean(message.pinned_at) === payload.pinned;
+        if (!already && payload.pinned && (await countPinnedMessages(payload.conversationId)) >= MAX_PINS) {
+          socket.emit("chat:error", { error: "too_many_pins", message: `A conversation can have ${MAX_PINS} pins. Unpin one first.` });
+          return;
+        }
+
+        const updated = already
+          ? message
+          : await setMessagePinned(payload.conversationId, payload.messageId, payload.pinned ? auth.tokenPayload.serverUserId : null);
+        if (!updated) { socket.emit("chat:error", "Failed to pin message"); return; }
+        replaceCachedMessage(payload.conversationId, updated);
+
+        const event = {
+          conversation_id: payload.conversationId,
+          message_id: payload.messageId,
+          pinned_at: updated.pinned_at ?? null,
+          pinned_by: updated.pinned_by ?? null,
+        };
+        const connectedClients = await recipientClientIds(payload.conversationId, access);
+        connectedClients.forEach((cid) => {
+          io.sockets.sockets.get(cid)?.emit("chat:pinned", event);
+        });
+      } catch (err) {
+        consola.error("chat:pin failed", err);
+        socket.emit("chat:error", "Failed to pin message");
+      }
+    },
+
+    // The pinned messages of a conversation, newest pin first, for the pins list.
+    'chat:pins': async (payload: { conversationId: string }) => {
+      try {
+        const ip = getClientIp();
+        const userId = clientsInfo[clientId]?.serverUserId;
+        if (!(await socketMay(clientsInfo, clientId, "read_messages"))) {
+          socket.emit("chat:error", { error: "forbidden", message: "You do not have permission to read this channel.", permission: "read_messages" });
+          return;
+        }
+        const rl = checkRateLimit("chat:fetch", userId, ip, RL_FETCH);
+        if (!rl.allowed) {
+          socket.emit("chat:error", { error: "rate_limited", retryAfterMs: rl.retryAfterMs, message: `Too fast. Wait ${Math.ceil((rl.retryAfterMs || 0) / 1000)}s.` });
+          return;
+        }
+        if (!payload || typeof payload.conversationId !== "string") { socket.emit("chat:error", "Invalid pins payload"); return; }
+        if (!(await requireConversationAccess(payload.conversationId, userId))) return;
+
+        const hidden = await blockedServerIdsFor(userId ?? "");
+        const pinned = (await listPinnedMessages(payload.conversationId, MAX_PINS)).filter((m) => !hidden.has(m.sender_server_id));
+        const items = await enrichAttachments(await enrichMessages(pinned));
+        socket.emit("chat:pins", { conversation_id: payload.conversationId, items });
+      } catch (err) {
+        consola.error("chat:pins failed", err);
+        socket.emit("chat:error", "Failed to fetch pins");
+      }
+    },
   };
 }