logos-messaging/js-waku
1
0
Fork 0
mirror of https://github.com/logos-messaging/js-waku.git synced 2026-01-09 09:13:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: abstract messageIndex behind ILocalHistory as findMissingDependencies

Browse Source
This commit is contained in:
Danish Arora 2025-11-28 10:56:34 -05:00
parent 055d7fecca
commit 42bfd98861
No known key found for this signature in database
GPG Key ID: 1C6EF37CDAE1426E
2 changed files with 26 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
packages/sds/src/message_channel/mem_local_history.ts
View File

@ -1,7 +1,12 @@
import { Logger } from "@waku/utils";
import _ from "lodash";
import { type ChannelId, ContentMessage, isContentMessage } from "./message.js";
import {
type ChannelId,
ContentMessage,
type HistoryEntry,
isContentMessage
} from "./message.js";
import { PersistentStorage } from "./persistent_storage.js";
export const DEFAULT_MAX_LENGTH = 10_000;
@ -26,6 +31,7 @@ export interface ILocalHistory {
getMessage(messageId: string): ContentMessage | undefined;
getRecentMessages(count: number): ContentMessage[];
getAllMessages(): ContentMessage[];
findMissingDependencies(entries: HistoryEntry[]): HistoryEntry[];
}
export type MemLocalHistoryOptions = {
@ -114,6 +120,10 @@ export class MemLocalHistory implements ILocalHistory {
return this.messageIndex.get(messageId);
}
public findMissingDependencies(entries: HistoryEntry[]): HistoryEntry[] {
return entries.filter((entry) => !this.messageIndex.has(entry.messageId));
}
private rebuildIndex(): void {
this.messageIndex.clear();
for (const message of this.items) {

21
packages/sds/src/message_channel/message_channel.ts
View File

@ -297,9 +297,8 @@ export class MessageChannel extends TypedEventEmitter<MessageChannelEvents> {
message.messageId,
message.causalHistory.map((ch) => ch.messageId)
);
const missingDependencies = message.causalHistory.filter(
(messageHistoryEntry) =>
!this.isMessageAvailable(messageHistoryEntry.messageId)
const missingDependencies = this.findMissingDependencies(
message.causalHistory
);
if (missingDependencies.length === 0) {
if (isContentMessage(message) && this.deliverMessage(message)) {
@ -583,9 +582,8 @@ export class MessageChannel extends TypedEventEmitter<MessageChannelEvents> {
this.filter.insert(message.messageId);
}
const missingDependencies = message.causalHistory.filter(
(messageHistoryEntry) =>
!this.isMessageAvailable(messageHistoryEntry.messageId)
const missingDependencies = this.findMissingDependencies(
message.causalHistory
);
if (missingDependencies.length > 0) {
@ -739,6 +737,17 @@ export class MessageChannel extends TypedEventEmitter<MessageChannelEvents> {
}
}
private findMissingDependencies(entries: HistoryEntry[]): HistoryEntry[] {
const missingFromHistory =
this.localHistory.findMissingDependencies(entries);
const incomingIds = new Set(this.incomingBuffer.map((m) => m.messageId));
return missingFromHistory.filter(
(entry) => !incomingIds.has(entry.messageId)
);
}
/**
* Check if a message is available (either in localHistory or incomingBuffer)
* This prevents treating messages as "missing" when they've already been received