fix(broker): drained pushes carry sender identity + topic like live pushes (bug2)

maybePushQueuedMessages sent queued rows with only senderPubkey, so the
daemon labelled them with an 8-char pubkey, couldn't tell a topic post
from a DM, and couldn't run its own-member echo filter. drainForMember
now returns senderMemberPubkey, senderName (session presence name, else
the member's) and the topic name for `#<topicId>` targets; the push
envelope forwards them plus senderMemberId.

Covered by two new broker.test.ts cases (run against pgvector pg16).
Spec: .artifacts/specs/2026-09-26-session-identity-and-routing.md §1

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Alejandro Gutiérrez
2026-09-26 20:54:23 +01:00
parent b5bb19c59c
commit 52e14d403b
3 changed files with 86 additions and 2 deletions

View File

@@ -2456,6 +2456,10 @@ export async function drainForMember(
/** v0.9.0 daemon fields; null for legacy traffic. */
clientMessageId: string | null;
requestFingerprint: string | null;
/** 1.38.0: identity + topic, mirroring the live push envelope. */
senderMemberPubkey: string;
senderName: string | null;
topic: string | null;
}>
> {
const priorities = deliverablePriorities(status);
@@ -2524,6 +2528,9 @@ export async function drainForMember(
sender_pubkey: string;
client_message_id: string | null;
request_fingerprint: string | null;
sender_member_pubkey: string;
sender_name: string | null;
topic_name: string | null;
}>(sql`
WITH claimed AS (
UPDATE mesh.message_queue AS mq
@@ -2546,7 +2553,22 @@ export async function drainForMember(
RETURNING mq.id, mq.priority, mq.nonce, mq.ciphertext,
mq.created_at, mq.sender_member_id,
mq.client_message_id, mq.request_fingerprint,
COALESCE(mq.sender_session_pubkey, m.peer_pubkey) AS sender_pubkey
COALESCE(mq.sender_session_pubkey, m.peer_pubkey) AS sender_pubkey,
-- 1.38.0 (spec 2026-09-26 §1): same identity fields the live
-- push carries, so a drained message renders with the right
-- author + topic instead of a bare 8-char pubkey.
m.peer_pubkey AS sender_member_pubkey,
COALESCE(
(SELECT p.display_name FROM mesh.presence p
WHERE mq.sender_session_pubkey IS NOT NULL
AND p.session_pubkey = mq.sender_session_pubkey
AND p.display_name IS NOT NULL
ORDER BY p.connected_at DESC LIMIT 1),
m.display_name
) AS sender_name,
CASE WHEN mq.target_spec LIKE '#%'
THEN (SELECT t.name FROM mesh.topic t WHERE t.id = substring(mq.target_spec from 2))
END AS topic_name
)
SELECT * FROM claimed ORDER BY created_at ASC, id ASC
`);
@@ -2561,6 +2583,9 @@ export async function drainForMember(
sender_pubkey: string;
client_message_id: string | null;
request_fingerprint: string | null;
sender_member_pubkey: string;
sender_name: string | null;
topic_name: string | null;
}>;
if (!rows || rows.length === 0) return [];
return rows.map((r) => ({
@@ -2574,6 +2599,9 @@ export async function drainForMember(
senderPubkey: r.sender_pubkey,
clientMessageId: r.client_message_id ?? null,
requestFingerprint: r.request_fingerprint ?? null,
senderMemberPubkey: r.sender_member_pubkey,
senderName: r.sender_name ?? null,
topic: r.topic_name ?? null,
}));
}

View File

@@ -731,6 +731,10 @@ async function maybePushQueuedMessages(
messageId: m.id,
meshId: conn.meshId,
senderPubkey: m.senderPubkey,
senderMemberPubkey: m.senderMemberPubkey,
senderMemberId: m.senderMemberId,
...(m.senderName ? { senderName: m.senderName } : {}),
...(m.topic ? { topic: m.topic } : {}),
priority: m.priority,
nonce: m.nonce,
ciphertext: m.ciphertext,

View File

@@ -12,7 +12,7 @@
import { afterAll, afterEach, describe, expect, test } from "vitest";
import { and, eq } from "drizzle-orm";
import { db } from "../src/db";
import { presence, pendingStatus } from "@turbostarter/db/schema/mesh";
import { meshTopic, meshTopicMember, presence, pendingStatus } from "@turbostarter/db/schema/mesh";
import {
applyPendingHookStatus,
connectPresence,
@@ -395,6 +395,58 @@ describe("targetSpec routing", () => {
expect(drained[0]!.ciphertext).toBe("hi everyone");
});
// 1.38.0 (spec 2026-09-26 §1): a drained push carries the same identity
// + topic fields as a live push. Without them the daemon rendered
// «<8-char pubkey>: <binary>» and couldn't tell a topic post from a DM.
test("drained topic post carries topic name + sender identity", async () => {
m = await setupTestMesh("drain-identity");
const [t] = await db
.insert(meshTopic)
.values({ meshId: m.meshId, name: "compras-orq" })
.returning({ id: meshTopic.id });
await db.insert(meshTopicMember).values({ topicId: t!.id, memberId: m.peerB.memberId });
const sessionPk = "c".repeat(64);
await db.insert(presence).values({
memberId: m.peerA.memberId, sessionId: "s-a", sessionPubkey: sessionPk,
displayName: "Intra-Back-session-1", pid: 1, cwd: "/tmp",
});
await queueMessage({
meshId: m.meshId,
senderMemberId: m.peerA.memberId,
senderSessionPubkey: sessionPk,
targetSpec: `#${t!.id}`,
priority: "now",
nonce: "nt",
ciphertext: "sealed",
});
const drained = await drainForMember(m.meshId, m.peerB.memberId, m.peerB.pubkey, "idle");
expect(drained).toHaveLength(1);
expect(drained[0]).toMatchObject({
topic: "compras-orq",
senderPubkey: sessionPk,
senderMemberPubkey: m.peerA.pubkey,
senderName: "Intra-Back-session-1",
});
});
test("drained DM without session falls back to the member display name", async () => {
m = await setupTestMesh("drain-member-name");
await queueMessage({
meshId: m.meshId,
senderMemberId: m.peerA.memberId,
targetSpec: m.peerB.pubkey,
priority: "now",
nonce: "nd",
ciphertext: "dm",
});
const drained = await drainForMember(m.meshId, m.peerB.memberId, m.peerB.pubkey, "idle");
expect(drained[0]).toMatchObject({
topic: null,
senderMemberPubkey: m.peerA.pubkey,
senderName: "peer-a-drain-member-name",
});
});
test("pubkey mismatch → message not drained", async () => {
m = await setupTestMesh("pubkey-mismatch");
await queueMessage({