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