atagen

malice

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