From 30e8824e2851f6c34f17e8a69fd133dbc7dbd017 Mon Sep 17 00:00:00 2001 From: Dan Ditomaso Date: Thu, 23 Apr 2026 22:39:49 -0400 Subject: [PATCH] feat(sdk): add MessageRepository port + InMemoryMessageRepository Lays the persistence groundwork for the chat slice migration. - MessageRepository port defines paginated reads (loadRecent, loadBefore), atomic writes (append, appendBatch), state updates, and retention pruning. Conversation keyed by ConversationKey tagged union (channel | direct peer). - RetentionPolicy: maxPerBucket + olderThanMs knobs. Consumer decides. - InMemoryMessageRepository ships with SDK as default + test fixture. - ChatClient accepts { repository?, retention?, initialLoadLimit? }. Lazy-hydrates per conversation on first subscribe; writes through on every inbound message; prunes after append when retention policy is set. - ChatClient.loadOlder(conv, before, limit) for pagination UI. - ChatStore.prepend() for older-first inserts. Tests: 5 new cases for InMemoryMessageRepository (paginate, update state, retention). 25 SDK tests total, all green. Web build unchanged. Paves the way for @meshtastic/sdk-storage-sqlocal to implement this port against SQLite/OPFS in a follow-up. --- packages/sdk/mod.ts | 18 ++++- packages/sdk/src/core/client/MeshClient.ts | 5 +- packages/sdk/src/features/chat/ChatClient.ts | 71 +++++++++++++++-- .../features/chat/domain/MessageRepository.ts | 39 +++++++++ packages/sdk/src/features/chat/index.ts | 8 ++ .../InMemoryMessageRepository.test.ts | 63 +++++++++++++++ .../repositories/InMemoryMessageRepository.ts | 79 +++++++++++++++++++ .../sdk/src/features/chat/state/chatStore.ts | 9 +++ 8 files changed, 284 insertions(+), 8 deletions(-) create mode 100644 packages/sdk/src/features/chat/domain/MessageRepository.ts create mode 100644 packages/sdk/src/features/chat/infrastructure/repositories/InMemoryMessageRepository.test.ts create mode 100644 packages/sdk/src/features/chat/infrastructure/repositories/InMemoryMessageRepository.ts diff --git a/packages/sdk/mod.ts b/packages/sdk/mod.ts index 5c62d372..50bd6743 100644 --- a/packages/sdk/mod.ts +++ b/packages/sdk/mod.ts @@ -53,8 +53,22 @@ export { DeviceClient } from "./src/features/device/index.ts"; export type { Device } from "./src/features/device/index.ts"; export { ChatClient } from "./src/features/chat/index.ts"; -export type { Message, SendTextError, SendTextInput } from "./src/features/chat/index.ts"; -export { EmptyMessageError, MessageState, MessageTooLongError } from "./src/features/chat/index.ts"; +export type { + ChatClientOptions, + ConversationKey, + Message, + MessageRepository, + RetentionPolicy, + SendTextError, + SendTextInput, +} from "./src/features/chat/index.ts"; +export { + conversationKeyString, + EmptyMessageError, + InMemoryMessageRepository, + MessageState, + MessageTooLongError, +} from "./src/features/chat/index.ts"; export { NodesClient } from "./src/features/nodes/index.ts"; export type { Node } from "./src/features/nodes/index.ts"; diff --git a/packages/sdk/src/core/client/MeshClient.ts b/packages/sdk/src/core/client/MeshClient.ts index 33c75869..6c7d2769 100644 --- a/packages/sdk/src/core/client/MeshClient.ts +++ b/packages/sdk/src/core/client/MeshClient.ts @@ -22,10 +22,13 @@ import { PositionClient } from "../../features/position/index.ts"; import { TelemetryClient } from "../../features/telemetry/index.ts"; import { TraceRouteClient } from "../../features/traceroute/index.ts"; +import type { ChatClientOptions } from "../../features/chat/ChatClient.ts"; + export interface MeshClientOptions { transport: Transport; configId?: number; logger?: Logger; + chat?: ChatClientOptions; } /** @@ -64,7 +67,7 @@ export class MeshClient { this.configId = options.configId ?? generatePacketId(); this.device = new DeviceClient(this); - this.chat = new ChatClient(this); + this.chat = new ChatClient(this, options.chat); this.nodes = new NodesClient(this); this.channels = new ChannelsClient(this); this.config = new ConfigClient(this); diff --git a/packages/sdk/src/features/chat/ChatClient.ts b/packages/sdk/src/features/chat/ChatClient.ts index 6d76bff1..65db5fba 100644 --- a/packages/sdk/src/features/chat/ChatClient.ts +++ b/packages/sdk/src/features/chat/ChatClient.ts @@ -4,49 +4,110 @@ import { Constants } from "../../core/constants/index.ts"; import type { ReadonlySignal } from "../../core/signals/createStore.ts"; import type { ChannelNumber } from "../../core/types.ts"; import type { Message } from "./domain/Message.ts"; +import type { + ConversationKey, + MessageRepository, + RetentionPolicy, +} from "./domain/MessageRepository.ts"; import { MessageState } from "./domain/MessageState.ts"; import { MessageMapper } from "./infrastructure/MessageMapper.ts"; +import { InMemoryMessageRepository } from "./infrastructure/repositories/InMemoryMessageRepository.ts"; import { type SendTextError, type SendTextInput, sendText } from "./application/SendTextUseCase.ts"; import { ChatStore } from "./state/chatStore.ts"; +export interface ChatClientOptions { + repository?: MessageRepository; + retention?: RetentionPolicy; + /** Messages to load into the store on first subscription of a conversation. */ + initialLoadLimit?: number; +} + /** * Chat slice facade. Exposes message buckets keyed by channel or peer, and the - * `send` command for outbound text. + * `send` command for outbound text. Optional persistence via MessageRepository. */ export class ChatClient { private readonly client: MeshClient; private readonly store: ChatStore; + private readonly repository: MessageRepository; + private readonly retention: RetentionPolicy | undefined; + private readonly initialLoadLimit: number; + private readonly hydrated = new Set(); - constructor(client: MeshClient) { + constructor(client: MeshClient, options: ChatClientOptions = {}) { this.client = client; this.store = new ChatStore(); + this.repository = options.repository ?? new InMemoryMessageRepository(); + this.retention = options.retention; + this.initialLoadLimit = options.initialLoadLimit ?? 50; client.events.onMessagePacket.subscribe((packet) => { const message = MessageMapper.fromPacket(packet); - const key = + const conv: ConversationKey = packet.type === "direct" && packet.to !== Constants.broadcastNum - ? this.store.directKey(packet.from === client.myNodeNum ? packet.to : packet.from) - : this.store.channelKey(packet.channel); + ? { kind: "direct", peer: packet.from === client.myNodeNum ? packet.to : packet.from } + : { kind: "channel", channel: packet.channel }; + const key = this.keyFor(conv); this.store.append(key, message); + void this.persistAppend(message); }); client.events.onRoutingPacket.subscribe((packet) => { if (packet.data.variant.case === "errorReason") { const state = packet.data.variant.value === 0 ? MessageState.Ack : MessageState.Failed; this.store.updateState(packet.id, state); + void this.repository.updateState(packet.id, state).catch(() => {}); } }); } public messages(channel: ChannelNumber): ReadonlySignal { + this.ensureHydrated({ kind: "channel", channel }); return this.store.messagesForChannel(channel); } public direct(peer: number): ReadonlySignal { + this.ensureHydrated({ kind: "direct", peer }); return this.store.messagesForDirect(peer); } + public async loadOlder(conv: ConversationKey, before: Date, limit = 50): Promise { + const older = await this.repository.loadBefore(conv, before, limit); + const key = this.keyFor(conv); + for (const m of older) this.store.prepend(key, m); + return older; + } + public send(input: SendTextInput): Promise> { return sendText(this.client, input); } + + private ensureHydrated(conv: ConversationKey): void { + const key = this.keyFor(conv); + if (this.hydrated.has(key)) return; + this.hydrated.add(key); + void (async () => { + try { + const recent = await this.repository.loadRecent(conv, this.initialLoadLimit); + for (const m of recent) this.store.append(key, m); + } catch { + // adapter may not have history yet; safe to ignore + } + })(); + } + + private async persistAppend(message: Message): Promise { + try { + await this.repository.append(message); + if (this.retention) await this.repository.prune(this.retention); + } catch { + // persistence failure must not break reactive flow + } + } + + private keyFor(conv: ConversationKey): string { + return conv.kind === "channel" + ? this.store.channelKey(conv.channel) + : this.store.directKey(conv.peer); + } } diff --git a/packages/sdk/src/features/chat/domain/MessageRepository.ts b/packages/sdk/src/features/chat/domain/MessageRepository.ts new file mode 100644 index 00000000..d34cfd1f --- /dev/null +++ b/packages/sdk/src/features/chat/domain/MessageRepository.ts @@ -0,0 +1,39 @@ +import type { ChannelNumber } from "../../../core/types.ts"; +import type { Message } from "./Message.ts"; +import type { MessageState } from "./MessageState.ts"; + +/** + * Conversation key. Broadcast messages are keyed by channel; direct messages + * are keyed by the peer node number. + */ +export type ConversationKey = + | { kind: "channel"; channel: ChannelNumber } + | { kind: "direct"; peer: number }; + +export interface RetentionPolicy { + /** Drop anything older than this many ms. */ + olderThanMs?: number; + /** Keep at most this many messages per conversation. */ + maxPerBucket?: number; +} + +/** + * Port for persisting chat messages. Implementations live in adapter packages + * (e.g. `@meshtastic/sdk-storage-sqlocal`) or in-memory within the SDK itself. + * + * Reads are paginated so consumers can lazy-load history on scroll rather than + * rehydrating every message at boot. + */ +export interface MessageRepository { + loadRecent(key: ConversationKey, limit: number): Promise; + loadBefore(key: ConversationKey, cursor: Date, limit: number): Promise; + append(message: Message): Promise; + appendBatch(messages: ReadonlyArray): Promise; + updateState(id: number, state: MessageState): Promise; + prune(policy: RetentionPolicy): Promise; + clear(): Promise; +} + +export function conversationKeyString(key: ConversationKey): string { + return key.kind === "channel" ? `channel:${key.channel}` : `direct:${key.peer}`; +} diff --git a/packages/sdk/src/features/chat/index.ts b/packages/sdk/src/features/chat/index.ts index f29b450d..d83ecd81 100644 --- a/packages/sdk/src/features/chat/index.ts +++ b/packages/sdk/src/features/chat/index.ts @@ -1,7 +1,15 @@ export { ChatClient } from "./ChatClient.ts"; +export type { ChatClientOptions } from "./ChatClient.ts"; export type { Message } from "./domain/Message.ts"; export { MessageState } from "./domain/MessageState.ts"; +export type { + ConversationKey, + MessageRepository, + RetentionPolicy, +} from "./domain/MessageRepository.ts"; +export { conversationKeyString } from "./domain/MessageRepository.ts"; export { MessageMapper } from "./infrastructure/MessageMapper.ts"; +export { InMemoryMessageRepository } from "./infrastructure/repositories/InMemoryMessageRepository.ts"; export { EmptyMessageError, MessageTooLongError, diff --git a/packages/sdk/src/features/chat/infrastructure/repositories/InMemoryMessageRepository.test.ts b/packages/sdk/src/features/chat/infrastructure/repositories/InMemoryMessageRepository.test.ts new file mode 100644 index 00000000..40fb3deb --- /dev/null +++ b/packages/sdk/src/features/chat/infrastructure/repositories/InMemoryMessageRepository.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, it } from "vitest"; +import { ChannelNumber } from "../../../../core/types.ts"; +import type { Message } from "../../domain/Message.ts"; +import { MessageState } from "../../domain/MessageState.ts"; +import { InMemoryMessageRepository } from "./InMemoryMessageRepository.ts"; + +function msg(id: number, ms: number, text = "t"): Message { + return { + id, + from: 1, + to: 0xffffffff, + channel: ChannelNumber.Primary, + rxTime: new Date(ms), + type: "broadcast", + text, + state: MessageState.Ack, + }; +} + +describe("InMemoryMessageRepository", () => { + it("loadRecent returns the tail of a bucket", async () => { + const repo = new InMemoryMessageRepository(); + await repo.appendBatch([msg(1, 1000), msg(2, 2000), msg(3, 3000)]); + const out = await repo.loadRecent({ kind: "channel", channel: ChannelNumber.Primary }, 2); + expect(out.map((m) => m.id)).toEqual([2, 3]); + }); + + it("loadBefore paginates older messages", async () => { + const repo = new InMemoryMessageRepository(); + await repo.appendBatch([msg(1, 1000), msg(2, 2000), msg(3, 3000), msg(4, 4000)]); + const out = await repo.loadBefore( + { kind: "channel", channel: ChannelNumber.Primary }, + new Date(3000), + 10, + ); + expect(out.map((m) => m.id)).toEqual([1, 2]); + }); + + it("updateState mutates the matching message", async () => { + const repo = new InMemoryMessageRepository(); + await repo.append(msg(42, 1000)); + await repo.updateState(42, MessageState.Failed); + const [found] = await repo.loadRecent({ kind: "channel", channel: ChannelNumber.Primary }, 1); + expect(found?.state).toBe(MessageState.Failed); + }); + + it("prune enforces maxPerBucket", async () => { + const repo = new InMemoryMessageRepository(); + await repo.appendBatch([msg(1, 1000), msg(2, 2000), msg(3, 3000), msg(4, 4000)]); + await repo.prune({ maxPerBucket: 2 }); + const out = await repo.loadRecent({ kind: "channel", channel: ChannelNumber.Primary }, 10); + expect(out.map((m) => m.id)).toEqual([3, 4]); + }); + + it("prune enforces olderThanMs", async () => { + const repo = new InMemoryMessageRepository(); + const now = Date.now(); + await repo.appendBatch([msg(1, now - 1000 * 60 * 60 * 24 * 10), msg(2, now)]); + await repo.prune({ olderThanMs: 1000 * 60 * 60 * 24 }); + const out = await repo.loadRecent({ kind: "channel", channel: ChannelNumber.Primary }, 10); + expect(out.map((m) => m.id)).toEqual([2]); + }); +}); diff --git a/packages/sdk/src/features/chat/infrastructure/repositories/InMemoryMessageRepository.ts b/packages/sdk/src/features/chat/infrastructure/repositories/InMemoryMessageRepository.ts new file mode 100644 index 00000000..ad015592 --- /dev/null +++ b/packages/sdk/src/features/chat/infrastructure/repositories/InMemoryMessageRepository.ts @@ -0,0 +1,79 @@ +import type { Message } from "../../domain/Message.ts"; +import type { MessageState } from "../../domain/MessageState.ts"; +import { + type ConversationKey, + conversationKeyString, + type MessageRepository, + type RetentionPolicy, +} from "../../domain/MessageRepository.ts"; + +/** + * Default in-memory MessageRepository. No persistence across reloads; useful + * for tests and for single-session apps that do not need history. + */ +export class InMemoryMessageRepository implements MessageRepository { + private readonly buckets = new Map(); + + async loadRecent(key: ConversationKey, limit: number): Promise { + const bucket = this.buckets.get(conversationKeyString(key)) ?? []; + return bucket.slice(-limit); + } + + async loadBefore(key: ConversationKey, cursor: Date, limit: number): Promise { + const bucket = this.buckets.get(conversationKeyString(key)) ?? []; + const idx = bucket.findIndex((m) => m.rxTime >= cursor); + const end = idx === -1 ? bucket.length : idx; + const start = Math.max(0, end - limit); + return bucket.slice(start, end); + } + + async append(message: Message): Promise { + await this.appendBatch([message]); + } + + async appendBatch(messages: ReadonlyArray): Promise { + for (const message of messages) { + const k = this.inferKey(message); + const bucket = this.buckets.get(k) ?? []; + bucket.push(message); + bucket.sort((a, b) => a.rxTime.getTime() - b.rxTime.getTime()); + this.buckets.set(k, bucket); + } + } + + async updateState(id: number, state: MessageState): Promise { + for (const bucket of this.buckets.values()) { + const idx = bucket.findIndex((m) => m.id === id); + if (idx !== -1) { + const existing = bucket[idx]; + if (!existing) continue; + bucket[idx] = { ...existing, state }; + return; + } + } + } + + async prune(policy: RetentionPolicy): Promise { + const nowMs = Date.now(); + for (const [key, bucket] of this.buckets) { + let filtered = + policy.olderThanMs === undefined + ? bucket + : bucket.filter((m) => nowMs - m.rxTime.getTime() <= policy.olderThanMs!); + if (policy.maxPerBucket !== undefined && filtered.length > policy.maxPerBucket) { + filtered = filtered.slice(-policy.maxPerBucket); + } + this.buckets.set(key, filtered); + } + } + + async clear(): Promise { + this.buckets.clear(); + } + + private inferKey(message: Message): string { + return message.type === "direct" + ? conversationKeyString({ kind: "direct", peer: message.from }) + : conversationKeyString({ kind: "channel", channel: message.channel }); + } +} diff --git a/packages/sdk/src/features/chat/state/chatStore.ts b/packages/sdk/src/features/chat/state/chatStore.ts index 878dd6ff..e0e96d2f 100644 --- a/packages/sdk/src/features/chat/state/chatStore.ts +++ b/packages/sdk/src/features/chat/state/chatStore.ts @@ -34,6 +34,15 @@ export class ChatStore { bucket.value = [...bucket.value, message]; } + /** + * Inserts an older message at the front of the bucket. Used when paginating + * backwards; preserves chronological order because callers feed older-first. + */ + prepend(key: string, message: Message): void { + const bucket = this.writeBucket(key); + bucket.value = [message, ...bucket.value]; + } + updateState(id: number, state: MessageState): void { for (const [, bucket] of this.buckets) { const idx = bucket.value.findIndex((m) => m.id === id);