moon
/
ditto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(streaming): use policy instead of hand coding

This commit is contained in:
P. Reis 2024-05-14 21:44:19 -03:00
parent 4d342dff4a
commit 0383726663
1 changed files with 8 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/controllers/api/streaming.ts
View File

@ -9,7 +9,7 @@ import { bech32ToPubkey } from '@/utils.ts';
import { renderReblog, renderStatus } from '@/views/mastodon/statuses.ts'; import { renderReblog, renderStatus } from '@/views/mastodon/statuses.ts';
import { hydrateEvents } from '@/storages/hydrate.ts'; import { hydrateEvents } from '@/storages/hydrate.ts';
import { Storages } from '@/storages.ts'; import { Storages } from '@/storages.ts';
import { getTagSet } from '@/tags.ts'; import { MuteListPolicy } from '@/policies/MuteListPolicy.ts';
const debug = Debug('ditto:streaming'); const debug = Debug('ditto:streaming');
@ -34,12 +34,11 @@ const streamSchema = z.enum([
type Stream = z.infer<typeof streamSchema>; type Stream = z.infer<typeof streamSchema>;
const streamingController: AppController = async (c) => { const streamingController: AppController = (c) => {
const upgrade = c.req.header('upgrade'); const upgrade = c.req.header('upgrade');
const token = c.req.header('sec-websocket-protocol'); const token = c.req.header('sec-websocket-protocol');
const stream = streamSchema.optional().catch(undefined).parse(c.req.query('stream')); const stream = streamSchema.optional().catch(undefined).parse(c.req.query('stream'));
const controller = new AbortController(); const controller = new AbortController();
const signal = c.req.raw.signal;
if (upgrade?.toLowerCase() !== 'websocket') { if (upgrade?.toLowerCase() !== 'websocket') {
return c.text('Please use websocket protocol', 400); return c.text('Please use websocket protocol', 400);
@ -50,16 +49,6 @@ const streamingController: AppController = async (c) => {
return c.json({ error: 'Invalid access token' }, 401); return c.json({ error: 'Invalid access token' }, 401);
} }
const mutedUsersSet = new Set();
if (pubkey) {
const [mutedUsers] = await Storages.admin.query([{ authors: [pubkey], kinds: [10000], limit: 1 }], { signal });
if (mutedUsers) {
for (const pubkey of getTagSet(mutedUsers.tags, 'p')) {
mutedUsersSet.add(pubkey);
}
}
}
const { socket, response } = Deno.upgradeWebSocket(c.req.raw, { protocol: token }); const { socket, response } = Deno.upgradeWebSocket(c.req.raw, { protocol: token });
function send(name: string, payload: object) { function send(name: string, payload: object) {
@ -84,9 +73,13 @@ const streamingController: AppController = async (c) => {
if (msg[0] === 'EVENT') { if (msg[0] === 'EVENT') {
const event = msg[2]; const event = msg[2];
if (mutedUsersSet.has(event.pubkey)) { if (pubkey) {
const policy = new MuteListPolicy(pubkey, Storages.admin);
const ok = await policy.call(event);
if (ok[2] === false) {
continue; continue;
} }
}
await hydrateEvents({ await hydrateEvents({
events: [event], events: [event],