atagen

malice

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