atagen

malice

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