atagen

malice

public
Star0
Watch0
0380cd8c765cdd7735d5c259c26abdeef712e742
2
ssh://brewery@pls.lobotomise.me:33/atagen/malice.git
web/src/lib/pending.tsTypescript161 lines4.7 KB
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
});