| 1 | import { get, writable } from 'svelte/store'; |
| 2 | import { sendMessage } from './api'; |
| 3 | import { timelines } from './store'; |
| 4 | import type { ReplyReference, TimelineItem } from './types'; |
| 5 | |
| 6 | export interface PendingMessage { |
| 7 | id: string; |
| 8 | body: string; |
| 9 | replyTo: ReplyReference | null; |
| 10 | status: 'sending' | 'failed'; |
| 11 | error: string | null; |
| 12 | createdAt: number; |
| 13 | } |
| 14 | |
| 15 | export const pendingByRoom = writable<Record<string, PendingMessage[]>>({}); |
| 16 | |
| 17 | interface PendingRecord extends PendingMessage { |
| 18 | roomId: string; |
| 19 | knownIds: Set<string>; |
| 20 | } |
| 21 | |
| 22 | const records = new Map<string, PendingRecord>(); |
| 23 | const queues = new Map<string, Promise<void>>(); |
| 24 | let nextSeq = 1; |
| 25 | |
| 26 | function nextId(): string { |
| 27 | return `pending-${nextSeq++}`; |
| 28 | } |
| 29 | |
| 30 | function ownIdentities(items: TimelineItem[]): Set<string> { |
| 31 | const out = new Set<string>(); |
| 32 | for (const item of items) { |
| 33 | if (item.kind !== 'event' || !item.event.own) continue; |
| 34 | const id = item.event.transaction_id ?? item.event.event_id; |
| 35 | if (id) out.add(id); |
| 36 | } |
| 37 | return out; |
| 38 | } |
| 39 | |
| 40 | function toPublic(record: PendingRecord): PendingMessage { |
| 41 | return { id: record.id, body: record.body, replyTo: record.replyTo, status: record.status, error: record.error, createdAt: record.createdAt }; |
| 42 | } |
| 43 | |
| 44 | function roomList(roomId: string): PendingMessage[] { |
| 45 | return get(pendingByRoom)[roomId] ?? []; |
| 46 | } |
| 47 | |
| 48 | function setRoom(roomId: string, list: PendingMessage[]): void { |
| 49 | pendingByRoom.update((all) => ({ ...all, [roomId]: list })); |
| 50 | } |
| 51 | |
| 52 | function addRecord(record: PendingRecord): void { |
| 53 | records.set(record.id, record); |
| 54 | setRoom(record.roomId, [...roomList(record.roomId), toPublic(record)]); |
| 55 | } |
| 56 | |
| 57 | function updateRecord(id: string, patch: Partial<Pick<PendingRecord, 'status' | 'error'>>): void { |
| 58 | const record = records.get(id); |
| 59 | if (!record) return; |
| 60 | Object.assign(record, patch); |
| 61 | setRoom(record.roomId, roomList(record.roomId).map((p) => (p.id === id ? toPublic(record) : p))); |
| 62 | } |
| 63 | |
| 64 | function removeRecord(id: string): void { |
| 65 | const record = records.get(id); |
| 66 | if (!record) return; |
| 67 | records.delete(id); |
| 68 | setRoom(record.roomId, roomList(record.roomId).filter((p) => p.id !== id)); |
| 69 | } |
| 70 | |
| 71 | async function attemptSend(record: PendingRecord): Promise<void> { |
| 72 | if (!records.has(record.id)) return; |
| 73 | updateRecord(record.id, { status: 'sending', error: null }); |
| 74 | try { |
| 75 | await sendMessage(record.roomId, record.body, record.replyTo?.event_id ?? null); |
| 76 | } catch (e) { |
| 77 | updateRecord(record.id, { status: 'failed', error: e instanceof Error ? e.message : 'Message could not be sent' }); |
| 78 | } |
| 79 | } |
| 80 | |
| 81 | function enqueue(record: PendingRecord): void { |
| 82 | const previous = queues.get(record.roomId) ?? Promise.resolve(); |
| 83 | const next = previous.then(() => attemptSend(record)); |
| 84 | queues.set(record.roomId, next.catch(() => undefined)); |
| 85 | } |
| 86 | |
| 87 | export function queueSend(roomId: string, body: string, replyTo: ReplyReference | null = null): void { |
| 88 | const record: PendingRecord = { |
| 89 | id: nextId(), |
| 90 | roomId, |
| 91 | body, |
| 92 | replyTo, |
| 93 | status: 'sending', |
| 94 | error: null, |
| 95 | createdAt: Date.now(), |
| 96 | knownIds: ownIdentities(get(timelines)[roomId] ?? []) |
| 97 | }; |
| 98 | addRecord(record); |
| 99 | enqueue(record); |
| 100 | } |
| 101 | |
| 102 | export function retryPending(id: string): void { |
| 103 | const record = records.get(id); |
| 104 | if (!record) return; |
| 105 | record.knownIds = ownIdentities(get(timelines)[record.roomId] ?? []); |
| 106 | enqueue(record); |
| 107 | } |
| 108 | |
| 109 | export function discardPending(id: string): void { |
| 110 | removeRecord(id); |
| 111 | } |
| 112 | |
| 113 | export function resetPending(): void { |
| 114 | records.clear(); |
| 115 | queues.clear(); |
| 116 | pendingByRoom.set({}); |
| 117 | } |
| 118 | |
| 119 | function reconcileRoom(roomId: string, items: TimelineItem[]): void { |
| 120 | const list = roomList(roomId); |
| 121 | if (list.length === 0) return; |
| 122 | const ownItems = items.filter((item) => item.kind === 'event' && item.event.own) as Extract< |
| 123 | TimelineItem, |
| 124 | { kind: 'event' } |
| 125 | >[]; |
| 126 | const claimed = new Set<string>(); |
| 127 | for (const pending of list) { |
| 128 | const record = records.get(pending.id); |
| 129 | if (!record) continue; |
| 130 | const match = ownItems.find((item) => { |
| 131 | if (claimed.has(item.event.item_id)) return false; |
| 132 | const id = item.event.transaction_id ?? item.event.event_id; |
| 133 | if (id && record.knownIds.has(id)) return false; |
| 134 | return item.event.content.kind === 'Message' && item.event.content.message.body === record.body; |
| 135 | }); |
| 136 | if (match) { |
| 137 | claimed.add(match.event.item_id); |
| 138 | removeRecord(record.id); |
| 139 | } |
| 140 | } |
| 141 | } |
| 142 | |
| 143 | timelines.subscribe((all) => { |
| 144 | for (const roomId of Object.keys(get(pendingByRoom))) { |
| 145 | reconcileRoom(roomId, all[roomId] ?? []); |
| 146 | } |
| 147 | }); |