fix(api,broker,cli): topic posts are labelled by the session that wrote them (bug6)
`topic tail` showed the member name («Intra-Back») for posts from three different sessions: REST posts carried no session identity, and every read path resolved the author through the member table only. - api POST /v1/messages accepts sessionPubkey and stores it on topic_message + message_queue — only when a presence row proves the key's member owns that session. - api /messages + /stream and broker topicHistory return COALESCE(session pubkey, member pubkey) and the session's latest presence name, falling back to the member's. - cli `topic post` sends the current session's pubkey. No migration: topic_message.sender_session_pubkey already existed. Broker test added (pgvector pg16); the api SQL was run against the same DB. Spec: .artifacts/specs/2026-09-26-session-identity-and-routing.md §6 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -1110,7 +1110,15 @@ export async function topicHistory(args: {
|
|||||||
}>(sql`
|
}>(sql`
|
||||||
SELECT tm.id, tm.sender_member_id,
|
SELECT tm.id, tm.sender_member_id,
|
||||||
COALESCE(tm.sender_session_pubkey, m.peer_pubkey) AS sender_pubkey,
|
COALESCE(tm.sender_session_pubkey, m.peer_pubkey) AS sender_pubkey,
|
||||||
m.display_name AS sender_name,
|
-- bug6 (2026-09-26): the writing session's name, else the member's.
|
||||||
|
COALESCE(
|
||||||
|
(SELECT p.display_name FROM mesh.presence p
|
||||||
|
WHERE tm.sender_session_pubkey IS NOT NULL
|
||||||
|
AND p.session_pubkey = tm.sender_session_pubkey
|
||||||
|
AND p.display_name IS NOT NULL
|
||||||
|
ORDER BY p.connected_at DESC LIMIT 1),
|
||||||
|
m.display_name
|
||||||
|
) AS sender_name,
|
||||||
tm.nonce, tm.ciphertext, tm.body_version, tm.reply_to_id,
|
tm.nonce, tm.ciphertext, tm.body_version, tm.reply_to_id,
|
||||||
tm.created_at
|
tm.created_at
|
||||||
FROM mesh.topic_message tm
|
FROM mesh.topic_message tm
|
||||||
|
|||||||
@@ -12,7 +12,7 @@
|
|||||||
import { afterAll, afterEach, describe, expect, test } from "vitest";
|
import { afterAll, afterEach, describe, expect, test } from "vitest";
|
||||||
import { and, eq } from "drizzle-orm";
|
import { and, eq } from "drizzle-orm";
|
||||||
import { db } from "../src/db";
|
import { db } from "../src/db";
|
||||||
import { meshTopic, meshTopicMember, presence, pendingStatus } from "@turbostarter/db/schema/mesh";
|
import { meshTopic, meshTopicMember, meshTopicMessage, presence, pendingStatus } from "@turbostarter/db/schema/mesh";
|
||||||
import {
|
import {
|
||||||
applyPendingHookStatus,
|
applyPendingHookStatus,
|
||||||
connectPresence,
|
connectPresence,
|
||||||
@@ -22,6 +22,7 @@ import {
|
|||||||
queueMessage,
|
queueMessage,
|
||||||
refreshStatusFromJsonl,
|
refreshStatusFromJsonl,
|
||||||
sweepStuckWorking,
|
sweepStuckWorking,
|
||||||
|
topicHistory,
|
||||||
writeStatus,
|
writeStatus,
|
||||||
} from "../src/broker";
|
} from "../src/broker";
|
||||||
import { cleanupAllTestMeshes, setupTestMesh, type TestMesh } from "./helpers";
|
import { cleanupAllTestMeshes, setupTestMesh, type TestMesh } from "./helpers";
|
||||||
@@ -447,6 +448,26 @@ describe("targetSpec routing", () => {
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// bug6 (2026-09-26): history labels each post with the session that
|
||||||
|
// wrote it — three sessions of one member all read «Intra-Back» before.
|
||||||
|
test("topic history labels posts by writing session, member as fallback", async () => {
|
||||||
|
m = await setupTestMesh("history-session-name");
|
||||||
|
const [t] = await db.insert(meshTopic).values({ meshId: m.meshId, name: "orq" }).returning({ id: meshTopic.id });
|
||||||
|
const s1 = "d".repeat(64);
|
||||||
|
await db.insert(presence).values({
|
||||||
|
memberId: m.peerA.memberId, sessionId: "s-1", sessionPubkey: s1,
|
||||||
|
displayName: "front-session", pid: 1, cwd: "/tmp",
|
||||||
|
});
|
||||||
|
await db.insert(meshTopicMessage).values([
|
||||||
|
{ topicId: t!.id, senderMemberId: m.peerA.memberId, senderSessionPubkey: s1, nonce: "n", ciphertext: "c1" },
|
||||||
|
{ topicId: t!.id, senderMemberId: m.peerA.memberId, nonce: "n", ciphertext: "c2" },
|
||||||
|
]);
|
||||||
|
const rows = await topicHistory({ topicId: t!.id });
|
||||||
|
const byCt = Object.fromEntries(rows.map((r) => [r.ciphertext, r]));
|
||||||
|
expect(byCt.c1).toMatchObject({ senderPubkey: s1, senderName: "front-session" });
|
||||||
|
expect(byCt.c2).toMatchObject({ senderPubkey: m.peerA.pubkey, senderName: "peer-a-history-session-name" });
|
||||||
|
});
|
||||||
|
|
||||||
test("pubkey mismatch → message not drained", async () => {
|
test("pubkey mismatch → message not drained", async () => {
|
||||||
m = await setupTestMesh("pubkey-mismatch");
|
m = await setupTestMesh("pubkey-mismatch");
|
||||||
await queueMessage({
|
await queueMessage({
|
||||||
|
|||||||
@@ -134,6 +134,10 @@ export async function runTopicPost(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// bug6 (2026-09-26): label the post with THIS session, not the member.
|
||||||
|
// The API honours it only when the key's member owns that session.
|
||||||
|
const { getSessionInfo } = await import("~/services/session/resolve.js");
|
||||||
|
const sessionPubkey = (await getSessionInfo())?.presence?.sessionPubkey;
|
||||||
const result = await request<PostResponse>({
|
const result = await request<PostResponse>({
|
||||||
path: "/api/v1/messages",
|
path: "/api/v1/messages",
|
||||||
method: "POST",
|
method: "POST",
|
||||||
@@ -145,6 +149,7 @@ export async function runTopicPost(
|
|||||||
bodyVersion,
|
bodyVersion,
|
||||||
...(mentions.length > 0 ? { mentions } : {}),
|
...(mentions.length > 0 ? { mentions } : {}),
|
||||||
...(replyToId ? { replyToId } : {}),
|
...(replyToId ? { replyToId } : {}),
|
||||||
|
...(sessionPubkey ? { sessionPubkey } : {}),
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
@@ -54,6 +54,21 @@ import {
|
|||||||
|
|
||||||
type Env = { Variables: { apiKey: AuthedApiKey } };
|
type Env = { Variables: { apiKey: AuthedApiKey } };
|
||||||
|
|
||||||
|
/**
|
||||||
|
* bug6 (2026-09-26): a topic post is labelled with the SESSION that wrote
|
||||||
|
* it, not the member — three sessions of one member all showed as
|
||||||
|
* «Intra-Back». The session name comes from its latest presence row;
|
||||||
|
* member identity is the fallback (REST posts from the dashboard, old CLIs).
|
||||||
|
*/
|
||||||
|
const topicSenderPubkey = sql<string>`COALESCE(${meshTopicMessage.senderSessionPubkey}, ${meshMember.peerPubkey})`;
|
||||||
|
const topicSenderName = sql<string>`COALESCE(
|
||||||
|
(SELECT p.display_name FROM mesh.presence p
|
||||||
|
WHERE ${meshTopicMessage.senderSessionPubkey} IS NOT NULL
|
||||||
|
AND p.session_pubkey = ${meshTopicMessage.senderSessionPubkey}
|
||||||
|
AND p.display_name IS NOT NULL
|
||||||
|
ORDER BY p.connected_at DESC LIMIT 1),
|
||||||
|
${meshMember.displayName})`;
|
||||||
|
|
||||||
const sendMessageSchema = z.object({
|
const sendMessageSchema = z.object({
|
||||||
topic: z.string().min(1),
|
topic: z.string().min(1),
|
||||||
/** base64-encoded ciphertext; client encrypts before sending. */
|
/** base64-encoded ciphertext; client encrypts before sending. */
|
||||||
@@ -86,6 +101,12 @@ const sendMessageSchema = z.object({
|
|||||||
* drops the reference (treated as a top-level post).
|
* drops the reference (treated as a top-level post).
|
||||||
*/
|
*/
|
||||||
replyToId: z.string().min(1).max(128).optional(),
|
replyToId: z.string().min(1).max(128).optional(),
|
||||||
|
/**
|
||||||
|
* bug6 (2026-09-26): the posting session's pubkey. Only honoured when a
|
||||||
|
* presence row proves the key's member owns that session — a member can
|
||||||
|
* label a post with its own sessions, never someone else's.
|
||||||
|
*/
|
||||||
|
sessionPubkey: z.string().regex(/^[0-9a-f]{64}$/i).optional(),
|
||||||
});
|
});
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -182,6 +203,16 @@ export const v1Router = new Hono<Env>()
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let senderSessionPubkey: string | null = null;
|
||||||
|
if (body.sessionPubkey) {
|
||||||
|
const [owned] = await db
|
||||||
|
.select({ id: presence.id })
|
||||||
|
.from(presence)
|
||||||
|
.where(and(eq(presence.memberId, senderMemberId), eq(presence.sessionPubkey, body.sessionPubkey.toLowerCase())))
|
||||||
|
.limit(1);
|
||||||
|
if (owned) senderSessionPubkey = body.sessionPubkey.toLowerCase();
|
||||||
|
}
|
||||||
|
|
||||||
// Persist to history (topic_message) + ephemeral queue (message_queue).
|
// Persist to history (topic_message) + ephemeral queue (message_queue).
|
||||||
// Broker's drain loop picks up the queue entry and pushes to live peers.
|
// Broker's drain loop picks up the queue entry and pushes to live peers.
|
||||||
const [historyRow] = await db
|
const [historyRow] = await db
|
||||||
@@ -189,6 +220,7 @@ export const v1Router = new Hono<Env>()
|
|||||||
.values({
|
.values({
|
||||||
topicId: topic.id,
|
topicId: topic.id,
|
||||||
senderMemberId,
|
senderMemberId,
|
||||||
|
senderSessionPubkey,
|
||||||
nonce: body.nonce,
|
nonce: body.nonce,
|
||||||
ciphertext: body.ciphertext,
|
ciphertext: body.ciphertext,
|
||||||
bodyVersion: body.bodyVersion,
|
bodyVersion: body.bodyVersion,
|
||||||
@@ -201,6 +233,7 @@ export const v1Router = new Hono<Env>()
|
|||||||
.values({
|
.values({
|
||||||
meshId: key.meshId,
|
meshId: key.meshId,
|
||||||
senderMemberId,
|
senderMemberId,
|
||||||
|
senderSessionPubkey,
|
||||||
targetSpec: "#" + topic.id,
|
targetSpec: "#" + topic.id,
|
||||||
priority: body.priority,
|
priority: body.priority,
|
||||||
nonce: body.nonce,
|
nonce: body.nonce,
|
||||||
@@ -1477,8 +1510,8 @@ export const v1Router = new Hono<Env>()
|
|||||||
.select({
|
.select({
|
||||||
id: meshTopicMessage.id,
|
id: meshTopicMessage.id,
|
||||||
senderMemberId: meshTopicMessage.senderMemberId,
|
senderMemberId: meshTopicMessage.senderMemberId,
|
||||||
senderPubkey: meshMember.peerPubkey,
|
senderPubkey: topicSenderPubkey,
|
||||||
senderName: meshMember.displayName,
|
senderName: topicSenderName,
|
||||||
nonce: meshTopicMessage.nonce,
|
nonce: meshTopicMessage.nonce,
|
||||||
ciphertext: meshTopicMessage.ciphertext,
|
ciphertext: meshTopicMessage.ciphertext,
|
||||||
bodyVersion: meshTopicMessage.bodyVersion,
|
bodyVersion: meshTopicMessage.bodyVersion,
|
||||||
@@ -1581,8 +1614,8 @@ export const v1Router = new Hono<Env>()
|
|||||||
.select({
|
.select({
|
||||||
id: meshTopicMessage.id,
|
id: meshTopicMessage.id,
|
||||||
senderMemberId: meshTopicMessage.senderMemberId,
|
senderMemberId: meshTopicMessage.senderMemberId,
|
||||||
senderPubkey: meshMember.peerPubkey,
|
senderPubkey: topicSenderPubkey,
|
||||||
senderName: meshMember.displayName,
|
senderName: topicSenderName,
|
||||||
nonce: meshTopicMessage.nonce,
|
nonce: meshTopicMessage.nonce,
|
||||||
ciphertext: meshTopicMessage.ciphertext,
|
ciphertext: meshTopicMessage.ciphertext,
|
||||||
bodyVersion: meshTopicMessage.bodyVersion,
|
bodyVersion: meshTopicMessage.bodyVersion,
|
||||||
|
|||||||
Reference in New Issue
Block a user