git » sdk » commit 5ce8033

diff --git a/borogove/Chat.hx b/borogove/Chat.hx
index 76e3c51..a15531a 100644
--- a/borogove/Chat.hx
+++ b/borogove/Chat.hx
@@ -264,9 +264,12 @@ abstract class Chat extends EventEmitter {
 		@param message the ChatMessageBuilder to send
 		@param e2eePreference optional explicit End-to-End Encryption preference for this message. If Default, uses the client's preferred allowed E2EE method.
 	**/
-	abstract public function sendMessage(message:ChatMessageBuilder, e2eePreference:OutgoingE2EEPreference = Default):Void;
+	// Not abstract to work around https://github.com/HaxeFoundation/haxe/issues/11666
+	public function sendMessage(message:ChatMessageBuilder, e2eePreference:OutgoingE2EEPreference = Default):Void { }
 
-	abstract private function sendMessageStanza(stanza: Stanza, ?outboxItem: OutboxItem, e2eePreference: OutgoingE2EEPreference = Default):Void;
+	@:allow(borogove)
+	// Not abstract to work around https://github.com/HaxeFoundation/haxe/issues/11666
+	private function sendMessageStanza(stanza: Stanza, ?outboxItem: OutboxItem, e2eePreference: OutgoingE2EEPreference = Default):Void { }
 
 	/**
 		Signals that all messages up to and including this one have probably
@@ -1270,8 +1273,7 @@ class DirectChat extends Chat {
 		});
 	}
 
-	@HaxeCBridge.noemit // on superclass as abstract
-	public function sendMessage(message: ChatMessageBuilder, e2eePreference: OutgoingE2EEPreference = Default):Void {
+	override public function sendMessage(message: ChatMessageBuilder, e2eePreference: OutgoingE2EEPreference = Default):Void {
 		if (uiState == Invited) uiState = Open;
 		if (typingTimer != null) typingTimer.stop();
 		client.chatActivity(this);
@@ -1341,7 +1343,7 @@ class DirectChat extends Chat {
 		});
 	}
 
-	private function sendMessageStanza(stanza: Stanza, ?outboxItem: OutboxItem, e2eePreference: OutgoingE2EEPreference = Default) {
+	override private function sendMessageStanza(stanza: Stanza, ?outboxItem: OutboxItem, e2eePreference: OutgoingE2EEPreference = Default) {
 		if (stanza.name != "message") throw "Can only send message stanza this way";
 
 		if (outboxItem == null) outboxItem = outbox.newItem();
@@ -2339,8 +2341,7 @@ class Channel extends Chat {
 		});
 	}
 
-	@HaxeCBridge.noemit // on superclass as abstract
-	public function sendMessage(message:ChatMessageBuilder, e2eePreference:OutgoingE2EEPreference = Default):Void {
+	override public function sendMessage(message:ChatMessageBuilder, e2eePreference:OutgoingE2EEPreference = Default):Void {
 		if (uiState == Invited) uiState = Open;
 		if (typingTimer != null) typingTimer.stop();
 		client.chatActivity(this);
@@ -2422,7 +2423,7 @@ class Channel extends Chat {
 		return client.outgoingE2EEPreference[0];
 	}
 
-	private function sendMessageStanza(stanza: Stanza, ?outboxItem: OutboxItem, e2eePreference: OutgoingE2EEPreference = Default) {
+	override private function sendMessageStanza(stanza: Stanza, ?outboxItem: OutboxItem, e2eePreference: OutgoingE2EEPreference = Default) {
 		if (stanza.name != "message") throw "Can only send message stanza this way";
 
 		if (outboxItem == null) outboxItem = outbox.newItem();
diff --git a/borogove/Client.hx b/borogove/Client.hx
index d9a7fce..b5feb2a 100644
--- a/borogove/Client.hx
+++ b/borogove/Client.hx
@@ -1105,6 +1105,8 @@ class Client extends EventEmitter {
 					if (!channel.inSync) channel.join();
 				}
 				firstSync = false;
+				// Once everything settles down, check db for resends
+				haxe.Timer.delay(() -> checkForResends(), 5000);
 			}
 
 			stream.emitSMupdates = true;
@@ -1163,6 +1165,8 @@ class Client extends EventEmitter {
 						sendPresence();
 						joinAllChannels();
 					}
+					// Once everything settles down, check db for resends
+					haxe.Timer.delay(() -> checkForResends(), 5000);
 					channelPinger.startTimer();
 					this.trigger("status/online", {});
 					trace("SYNC: done");
@@ -2029,6 +2033,36 @@ class Client extends EventEmitter {
 		})).then(updatedMessages -> persistence.storeMessages(accountId(), updatedMessages));
 	}
 
+	private function checkForResends() {
+		var tooOld = Date.format(
+			DateTools.delta(std.Date.now(), DateTools.hours(-3))
+		);
+
+		persistence.getMessagesByStatus(accountId(), MessagePending).then(messages -> {
+			for (m in messages) {
+				if (m.timestamp < tooOld) {
+					persistence.updateMessageStatus(accountId(), m.localId, MessageFailedToSend, "Message too old to auto-resend").then(newM -> {
+						notifyMessageHandlers(newM, StatusEvent);
+					});
+				} else {
+					final chat = getChat(m.chatId());
+					if (chat == null) {
+						trace("Resend chat not found", m.chatId(), m);
+					} else {
+						final stanza = m.asStanza();
+						stanza.textTag("delay", "Resending after reconnect", {
+							xmlns: "urn:xmpp:delay",
+							from: accountId(),
+							stamp: m.timestamp
+						});
+						// TODO: don't know if was originally send with OMEMO
+						chat.sendMessageStanza(stanza);
+					}
+				}
+			}
+		});
+	}
+
 	@:allow(borogove)
 	private function sendQuery(query:GenericQuery) {
 		this.stream.sendIq(query.getQueryStanza(), query.handleResponse);
diff --git a/borogove/Persistence.hx b/borogove/Persistence.hx
index 257ab17..f1d9f66 100644
--- a/borogove/Persistence.hx
+++ b/borogove/Persistence.hx
@@ -211,6 +211,16 @@ interface Persistence {
 	**/
 	public function getMessagesAround(accountId: String, around: ChatMessage): Promise<Array<ChatMessage>>;
 
+	/**
+		Load messages with a particular delivery status
+
+		@param accountId the account to load messages for
+		@param status delivery status to match
+		@returns Promise resolving to matching messages, oldest first
+	**/
+	@HaxeCBridge.noemit
+	public function getMessagesByStatus(accountId: String, status:borogove.Message.MessageStatus): Promise<Array<ChatMessage>>;
+
 	/**
 		Check whether a media blob is already stored
 
diff --git a/borogove/persistence/Dummy.hx b/borogove/persistence/Dummy.hx
index 9c1bebd..64e7fd3 100644
--- a/borogove/persistence/Dummy.hx
+++ b/borogove/persistence/Dummy.hx
@@ -106,6 +106,11 @@ class Dummy implements Persistence {
 		return Promise.resolve([]);
 	}
 
+	@HaxeCBridge.noemit
+	public function getMessagesByStatus(accountId: String, status: MessageStatus): Promise<Array<ChatMessage>> {
+		return Promise.resolve([]);
+	}
+
 	@HaxeCBridge.noemit
 	public function getChatsUnreadDetails(accountId: String, chats: Array<Chat>): Promise<Array<{ chatId: String, message: ChatMessage, unreadCount: Int }>> {
 		return Promise.resolve([]);
diff --git a/borogove/persistence/IDB.js b/borogove/persistence/IDB.js
index 3f45d7e..62b140f 100644
--- a/borogove/persistence/IDB.js
+++ b/borogove/persistence/IDB.js
@@ -268,6 +268,13 @@ export default async (dbname, media, tokenize, stemmer) => {
 						"timestamp",
 					]);
 				}
+				if (!messagesIndexNames.contains("accountsByStatus")) {
+					tx.objectStore("messages").createIndex("accountsByStatus", [
+						"account",
+						"status",
+						"timestamp",
+					]);
+				}
 			};
 			dbOpenReq.onsuccess = (event) => {
 				const db = event.target.result;
@@ -299,6 +306,7 @@ export default async (dbname, media, tokenize, stemmer) => {
 					"accountsBySortId",
 					"terms",
 					"chats",
+					"accountsByStatus",
 				];
 				for (const indexName of wantIndexNames) {
 					if (!messagesIndexNames.contains(indexName)) {
@@ -1601,6 +1609,24 @@ export default async (dbname, media, tokenize, stemmer) => {
 			);
 		},
 
+		getMessagesByStatus: async function (account, status) {
+			const tx = db.transaction(["messages"], "readonly");
+			const cursor = tx
+				.objectStore("messages")
+				.index("accountsByStatus")
+				.openCursor(
+					IDBKeyRange.bound([account, status], [account, status, []]),
+				);
+			const messages = [];
+			while (true) {
+				const result = await promisifyRequest(cursor);
+				if (!result) break;
+				messages.push(hydrateMessage(result.value));
+				result.continue();
+			}
+			return Promise.all(messages);
+		},
+
 		getMessagesFromCursor: async function (cursor, notIncluding, filter) {
 			const result = [];
 			while (true) {
diff --git a/borogove/persistence/Sqlite.hx b/borogove/persistence/Sqlite.hx
index 8120bc3..d6220b9 100644
--- a/borogove/persistence/Sqlite.hx
+++ b/borogove/persistence/Sqlite.hx
@@ -368,6 +368,12 @@ class Sqlite implements Persistence implements KeyValueStore {
 							"PRAGMA user_version = 22"]);
 					}
 					return Promise.resolve(null);
+				}).then(_ -> {
+					if (version < 23) {
+						return exec(["CREATE INDEX messages_status ON messages (account_id, status, created_at)",
+							"PRAGMA user_version = 23"]);
+					}
+					return Promise.resolve(null);
 				});
 			});
 		});
@@ -1046,6 +1052,33 @@ class Sqlite implements Persistence implements KeyValueStore {
 		]).then(results -> results.flatten());
 	}
 
+	@HaxeCBridge.noemit
+	public function getMessagesByStatus(accountId: String, status: MessageStatus): Promise<Array<ChatMessage>> {
+		return db.exec(
+			'
+				SELECT
+					${messageColumnsString()}
+				FROM messages
+				WHERE account_id=? AND status=?
+				ORDER BY created_at
+			',
+			[accountId, status]
+		).then(result -> {
+			final messages = hydrateMessages(accountId, result);
+			final replyTos = [];
+			final attachments = [];
+			for (message in messages) {
+				if (message.replyToMessage != null && message.replyToMessage.stanza == null) {
+					replyTos.push({ chatId: message.chatId(), serverId: message.replyToMessage.serverId, localId: message.replyToMessage.localId });
+				}
+				for (attachment in message.attachments) {
+					attachments.push(attachment.lookup(this));
+				}
+			}
+			return thenshim.PromiseTools.all(attachments).then(_ -> hydrateReplyTo(accountId, messages, replyTos));
+		}).then(messages -> hydrateReactions(accountId, messages));
+	}
+
 	private function getChatUnreadDetails(accountId: String, chat: Chat): Promise<{ chatId: String, message: ChatMessage, unreadCount: Int }> {
 		return db.exec(
 			'
diff --git a/test/TestClient.hx b/test/TestClient.hx
index 340caeb..026b659 100644
--- a/test/TestClient.hx
+++ b/test/TestClient.hx
@@ -8,6 +8,7 @@ import borogove.Chat;
 import borogove.ChatMessage;
 import borogove.ChatMessageBuilder;
 import borogove.Client;
+import borogove.Date;
 import borogove.JID;
 import borogove.Member;
 import borogove.MemberUpdate;
@@ -118,6 +119,43 @@ class TestClient extends utest.Test {
 		Assert.equals("test@example.com", client.accountId());
 	}
 
+	@:timeout(3000)
+	public function testReconnectResendsRecentPendingMessages(async: Async) {
+		final persistence = new ResendMockPersistence();
+		final client = new Client("test@example.com", persistence);
+		client.getDirectChat("friend@example.com");
+		final now = std.Date.now();
+		persistence.messages = [
+			pendingMessage("recent", Date.format(now)),
+			pendingMessage("stale", Date.format(DateTools.delta(now, DateTools.hours(-4))))
+		];
+
+		client.stream.on("sendStanza", (stanza: Stanza) -> {
+			if (stanza.attr.get("id") == "recent") {
+				Assert.equals("Resending after reconnect", stanza?.getChildText("delay", "urn:xmpp:delay"));
+				Assert.equals(MessageFailedToSend, persistence.statusUpdates.get("stale"));
+				async.done();
+			}
+			return EventHandled;
+		});
+
+		client.checkForResends();
+	}
+
+	private function pendingMessage(localId: String, timestamp: String): ChatMessage {
+		final builder = new ChatMessageBuilder();
+		builder.localId = localId;
+		builder.timestamp = timestamp;
+		builder.senderId = "test@example.com";
+		builder.from = JID.parse("test@example.com");
+		builder.to = JID.parse("friend@example.com");
+		builder.recipients = [JID.parse("friend@example.com")];
+		builder.text = "Test message";
+		builder.direction = MessageSent;
+		builder.status = MessagePending;
+		return builder.build();
+	}
+
 	@:timeout(3000)
 	public function testPrepareAttachmentNoUploadService(async: Async) {
 		final client = new Client("test@example.com", new Dummy());
@@ -1135,6 +1173,26 @@ class MessageMockPersistence extends Dummy {
 	}
 }
 
+@:access(borogove)
+class ResendMockPersistence extends Dummy {
+	public var messages: Array<ChatMessage> = [];
+	public var statusUpdates: Map<String, MessageStatus> = [];
+
+	override public function getMessagesByStatus(accountId: String, status: MessageStatus): Promise<Array<ChatMessage>> {
+		return Promise.resolve(messages.filter(message -> message.status == status));
+	}
+
+	override public function updateMessageStatus(accountId: String, localId: String, status: MessageStatus, statusText: Null<String>): Promise<ChatMessage> {
+		final message = messages.find(message -> message.localId == localId);
+		Assert.notNull(message);
+		statusUpdates.set(localId, status);
+		final builder = ChatMessageBuilder.fromMessage(message);
+		builder.status = status;
+		builder.statusText = statusText;
+		return Promise.resolve(builder.build());
+	}
+}
+
 @:access(borogove)
 class MemberUpdateMockPersistence extends Dummy {
 	public var lastUpdates: Array<MemberUpdate> = [];
diff --git a/test/TestSqlite.hx b/test/TestSqlite.hx
index 6f7bd5f..7eb75ef 100644
--- a/test/TestSqlite.hx
+++ b/test/TestSqlite.hx
@@ -716,6 +716,45 @@ class TestSqlite extends utest.Test {
 		});
 	}
 
+	public function testGetMessagesByStatus(async: Async) {
+		final account = "alice@example.com";
+		final pendingOld = makeMessage({
+			timestamp: "2020-01-01T00:00:00Z",
+			localId: "pending-old",
+			status: MessagePending
+		});
+		final delivered = makeMessage({
+			timestamp: "2020-01-01T00:00:01Z",
+			localId: "delivered",
+			status: MessageDeliveredToServer
+		});
+		final pendingNew = makeMessage({
+			timestamp: "2020-01-01T00:00:02Z",
+			localId: "pending-new",
+			status: MessagePending
+		});
+		final otherAccountPending = makeMessage({
+			timestamp: "2020-01-01T00:00:03Z",
+			localId: "other-account-pending",
+			status: MessagePending
+		});
+
+		persistence.storeMessages(account, [pendingNew, delivered, pendingOld]).then(_ -> {
+			return persistence.storeMessages("other@example.com", [otherAccountPending]);
+		}).then(_ -> {
+			return persistence.getMessagesByStatus(account, MessagePending);
+		}).then(messages -> {
+			Assert.equals(2, messages.length);
+			Assert.equals("pending-old", messages[0].localId);
+			Assert.equals("pending-new", messages[1].localId);
+			Assert.equals(MessagePending, messages[0].status);
+			async.done();
+		}).catchError(e -> {
+			Assert.fail(Std.string(e));
+			async.done();
+		});
+	}
+
 	private function makeMessage(params: {
 		timestamp: String,
 		?localId: String,
@@ -730,7 +769,8 @@ class TestSqlite extends utest.Test {
 		?encryption: EncryptionInfo,
 		?source: String,
 		?body: String,
-		?received: Bool
+		?received: Bool,
+		?status: MessageStatus
 	}):ChatMessage {
 		final senderId = params.senderId ?? "version@example.com";
 		final chatId = params.chatId ?? "chat@example.com";
@@ -743,6 +783,7 @@ class TestSqlite extends utest.Test {
 		builder.sortId = params.sortId ?? params.localId ?? params.serverId ?? "message";
 		builder.timestamp = params.timestamp;
 		builder.syncPoint = params.syncPoint ?? false;
+		builder.status = params.status ?? MessagePending;
 		builder.versions = params.versions ?? [];
 		builder.encryption = params.encryption;
 		if (params.source != null) builder.debug = { source: params.source };
diff --git a/test/persistence-tests.ts b/test/persistence-tests.ts
index 66f908a..17016ed 100644
--- a/test/persistence-tests.ts
+++ b/test/persistence-tests.ts
@@ -1,6 +1,60 @@
 import { expect } from "vitest";
 
 export function sharedPersistenceTests(test) {
+	test("getMessagesByStatus filters and orders messages", async ({
+		borogove,
+		persistence,
+	}) => {
+		const account = "alice@example.com";
+		const makeMessage = (localId, timestamp, status) => {
+			const builder = new borogove.ChatMessageBuilder({
+				localId,
+				senderId: account,
+				direction: borogove.MessageDirection.MessageSent,
+			});
+			builder.timestamp = timestamp;
+			builder.sortId = localId;
+			builder.status = status;
+			builder.to = borogove.JID.parse("hatter@example.com");
+			builder.from = borogove.JID.parse(account);
+			builder.recipients = [builder.to];
+			builder.replyTo = [builder.from];
+			return builder.build();
+		};
+
+		await persistence.storeMessages(account, [
+			makeMessage(
+				"pending-new",
+				"2020-01-01T00:00:02Z",
+				borogove.MessageStatus.MessagePending,
+			),
+			makeMessage(
+				"delivered",
+				"2020-01-01T00:00:01Z",
+				borogove.MessageStatus.MessageDeliveredToServer,
+			),
+			makeMessage(
+				"pending-old",
+				"2020-01-01T00:00:00Z",
+				borogove.MessageStatus.MessagePending,
+			),
+		]);
+
+		const pending = await persistence.getMessagesByStatus(
+			account,
+			borogove.MessageStatus.MessagePending,
+		);
+		expect(pending.map((message) => message.localId)).toEqual([
+			"pending-old",
+			"pending-new",
+		]);
+		expect(
+			pending.every(
+				(message) => message.status === borogove.MessageStatus.MessagePending,
+			),
+		).toBe(true);
+	});
+
 	test("message debug metadata round trips", async ({
 		borogove,
 		persistence,