From 52e14d403b5f005d91b124c9da06475317ac69ee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alejandro=20Guti=C3=A9rrez?= <35082514+alezmad@users.noreply.github.com> Date: Sat, 26 Sep 2026 20:54:23 +0100 Subject: [PATCH] fix(broker): drained pushes carry sender identity + topic like live pushes (bug2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 `#` 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 --- apps/broker/src/broker.ts | 30 +++++++++++++++++- apps/broker/src/index.ts | 4 +++ apps/broker/tests/broker.test.ts | 54 +++++++++++++++++++++++++++++++- 3 files changed, 86 insertions(+), 2 deletions(-) diff --git a/apps/broker/src/broker.ts b/apps/broker/src/broker.ts index eaf130a..0115e3b 100644 --- a/apps/broker/src/broker.ts +++ b/apps/broker/src/broker.ts @@ -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, })); } diff --git a/apps/broker/src/index.ts b/apps/broker/src/index.ts index 22a1a10..905df07 100644 --- a/apps/broker/src/index.ts +++ b/apps/broker/src/index.ts @@ -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, diff --git a/apps/broker/tests/broker.test.ts b/apps/broker/tests/broker.test.ts index 23c1a80..f7f8bb6 100644 --- a/apps/broker/tests/broker.test.ts +++ b/apps/broker/tests/broker.test.ts @@ -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>: » 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({