atagen

malice

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