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
8 changes: 8 additions & 0 deletions src/db/sqlite/messages.ts
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,14 @@ export async function listMessages(conversationId: string, limit = 50, before?:
return messages;
}

/** Oldest-first from a point onwards: the newer half of a jump to an old message, and the pages
below it until the window reaches the present (GRYT-1686). */
export async function listMessagesAfter(conversationId: string, limit: number, after: Date): Promise<MessageRecord[]> {
const db = getSqliteDb();
const rows = db.prepare(`SELECT * FROM messages WHERE conversation_id = ? AND thread_id IS NULL AND created_at > ? ORDER BY created_at ASC, message_id ASC LIMIT ?`).all(conversationId, toIso(after), limit);
return (rows as Record<string, unknown>[]).map(rowToMessage);
}

/**
* Newest-first, reversed on the way out: ascending with a limit took the oldest
* N, so a thread past its page hid the new replies.
Expand Down
70 changes: 70 additions & 0 deletions src/db/sqlite/messagesWindow.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
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 { insertMessage, listMessages, listMessagesAfter } from "./messages";

/** The two halves of a jump to an old message (GRYT-1686): `listMessages` before it and
`listMessagesAfter` past it, which together must cover the channel with no gap or repeat. */

let dir: string;
const CONV = "channel-window";
const START = Date.UTC(2026, 9, 1, 8, 0, 0);
const at = (i: number) => new Date(START + i * 60_000);

before(async () => {
dir = mkdtempSync(join(tmpdir(), "gryt-window-"));
process.env.DATA_DIR = dir;
await initSqlite();
for (let i = 0; i < 30; i++) {
await insertMessage({ conversation_id: CONV, message_id: `m${String(i).padStart(2, "0")}`, sender_server_id: "u_a", text: `message ${i}`, attachments: null, reactions: null, created_at: at(i) });
}
// A thread reply in the middle, which neither half may return.
await insertMessage({ conversation_id: CONV, message_id: "reply", sender_server_id: "u_a", text: "reply", attachments: null, reactions: null, thread_id: "t1", created_at: new Date(at(12).getTime() + 1) });
});

after(() => {
rmSync(dir, { recursive: true, force: true });
});

describe("listMessagesAfter", () => {
it("returns the next messages oldest first, starting after the point", async () => {
const page = await listMessagesAfter(CONV, 5, at(10));
assert.deepEqual(page.map((m) => m.message_id), ["m11", "m12", "m13", "m14", "m15"]);
});

it("leaves out thread replies", async () => {
const page = await listMessagesAfter(CONV, 30, at(12));
assert.ok(!page.some((m) => m.message_id === "reply"));
});

it("comes back short at the newest message", async () => {
const page = await listMessagesAfter(CONV, 10, at(25));
assert.deepEqual(page.map((m) => m.message_id), ["m26", "m27", "m28", "m29"]);
});

it("joins with listMessages around a message without a gap or a repeat", async () => {
const older = await listMessages(CONV, 4, at(15));
const newer = await listMessagesAfter(CONV, 4, at(15));
assert.deepEqual(
[...older, ...newer].map((m) => m.message_id),
["m11", "m12", "m13", "m14", "m16", "m17", "m18", "m19"],
);
});

it("walks from the oldest message to the newest in pages", async () => {
const seen: string[] = [];
let cursor = new Date(START - 1);
for (;;) {
const page = await listMessagesAfter(CONV, 7, cursor);
seen.push(...page.map((m) => m.message_id));
if (page.length < 7) break;
cursor = page[page.length - 1].created_at;
}
assert.equal(seen.length, 30);
assert.equal(new Set(seen).size, 30);
});
});
60 changes: 52 additions & 8 deletions src/socket/handlers/chat.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import { socketMay } from "../utils/standing";
import {
insertMessage,
listMessages,
listMessagesAfter,
listServerChannels,
getServerChannel,
MessageRecord,
Expand Down Expand Up @@ -863,7 +864,7 @@ export function registerChatHandlers(ctx: HandlerContext): EventHandlerMap {
}
},

'chat:fetch': async (payload: { conversationId: string; limit?: number; before?: string }) => {
'chat:fetch': async (payload: { conversationId: string; limit?: number; before?: string; after?: string; around?: string }) => {
try {
const ip = getClientIp();
const userId = clientsInfo[clientId]?.serverUserId;
Expand Down Expand Up @@ -899,12 +900,46 @@ export function registerChatHandlers(ctx: HandlerContext): EventHandlerMap {
}
}

const limit = typeof payload.limit === "number" ? payload.limit : 50;
const before = typeof payload.before === "string" ? new Date(payload.before) : undefined;
consola.info("[chat:fetch]", { conversationId: payload.conversationId, limit, before: before?.toISOString(), hasBefore: !!before });
const items = before
? await listMessages(payload.conversationId, limit, before)
: await getMessagesCached(payload.conversationId, limit);
// Capped: one request could otherwise ask for a whole channel's history in a page.
const limit = Math.min(Math.max(typeof payload.limit === "number" ? Math.floor(payload.limit) : 50, 1), 100);
const dateOf = (v: unknown) => {
const d = typeof v === "string" ? new Date(v) : undefined;
return d && !Number.isNaN(d.getTime()) ? d : undefined;
};
const before = dateOf(payload.before);
const after = dateOf(payload.after);
const around = typeof payload.around === "string" ? payload.around : undefined;
consola.info("[chat:fetch]", { conversationId: payload.conversationId, limit, before: before?.toISOString(), after: after?.toISOString(), around });

/* `around` opens history at one message, half a page each side; `after` pages towards
the present from there. Both say whether newer pages remain (GRYT-1686). */
let items: MessageRecord[];
let hasNewer: boolean | undefined;
let hasOlder: boolean | undefined;
let anchorFound: boolean | undefined;
if (around) {
const anchor = await getMessageById(payload.conversationId, around);
anchorFound = !!anchor && !anchor.thread_id;
if (anchor && anchorFound) {
const half = Math.max(1, Math.floor(limit / 2));
const older = await listMessages(payload.conversationId, half, anchor.created_at);
const newer = await listMessagesAfter(payload.conversationId, half, anchor.created_at);
items = [...older, anchor, ...newer];
hasOlder = older.length >= half;
hasNewer = newer.length >= half;
} else {
items = [];
hasOlder = false;
hasNewer = false;
}
} else if (after) {
items = await listMessagesAfter(payload.conversationId, limit, after);
hasNewer = items.length >= limit;
} else {
items = before
? await listMessages(payload.conversationId, limit, before)
: await getMessagesCached(payload.conversationId, limit);
}
/* Before enrichment, so nothing is spent on an avatar nobody sees.
`hasMore` counts what is left, so a page may come back short. */
const hidden = await blockedServerIdsFor(clientsInfo[clientId]?.serverUserId ?? "");
Expand All @@ -928,12 +963,21 @@ export function registerChatHandlers(ctx: HandlerContext): EventHandlerMap {
threads: typeof threads;
hasMore: boolean;
before?: string;
after?: string;
around?: string;
anchorFound?: boolean;
hasNewer?: boolean;
} = {
conversation_id: payload.conversationId,
items: enrichedItems,
threads,
hasMore: enrichedItems.length >= limit,
// An `after` page sits above the present, so older ones are known to exist.
hasMore: hasOlder ?? (after ? true : enrichedItems.length >= limit),
};
// Only on the new requests, so an older client sees the shape it always did.
if (around) { response.around = around; response.anchorFound = anchorFound; }
if (after) response.after = payload.after;
if (hasNewer !== undefined) response.hasNewer = hasNewer;
consola.info("[chat:fetch] response", { itemCount: enrichedItems.length, hasMore: response.hasMore, before: response.before });
if (before) response.before = payload.before;
socket.emit("chat:history", response);
Expand Down
Loading