pipeline: broadcast deletions to all known relays

This commit is contained in:
Alex Gleason 2023-09-05 21:38:15 -05:00
parent e2b88d57d9
commit a69b7f54f8
No known key found for this signature in database
GPG Key ID: 7211D1F99744FBB7
2 changed files with 24 additions and 3 deletions

View File

@ -4,15 +4,15 @@ import { nostrNow } from '@/utils.ts';
import * as pipeline from './pipeline.ts'; import * as pipeline from './pipeline.ts';
const relays = await getActiveRelays(); const _relays = await getActiveRelays();
const pool = new RelayPool(relays); const pool = new RelayPool(_relays);
// This file watches events on all known relays and performs // This file watches events on all known relays and performs
// side-effects based on them, such as trending hashtag tracking // side-effects based on them, such as trending hashtag tracking
// and storing events for notifications and the home feed. // and storing events for notifications and the home feed.
pool.subscribe( pool.subscribe(
[{ kinds: [0, 1, 3, 5, 6, 7, 10002], limit: 0, since: nostrNow() }], [{ kinds: [0, 1, 3, 5, 6, 7, 10002], limit: 0, since: nostrNow() }],
relays, _relays,
handleEvent, handleEvent,
undefined, undefined,
undefined, undefined,
@ -26,3 +26,10 @@ function handleEvent(event: Event): Promise<void> {
.handleEvent(event) .handleEvent(event)
.catch(() => {}); .catch(() => {});
} }
/** Publish an event to the given relays, or the entire pool. */
function publish(event: Event, relays: string[] = _relays) {
return pool.publish(event, relays);
}
export { publish };

View File

@ -3,6 +3,7 @@ import * as eventsDB from '@/db/events.ts';
import { addRelays } from '@/db/relays.ts'; import { addRelays } from '@/db/relays.ts';
import { findUser } from '@/db/users.ts'; import { findUser } from '@/db/users.ts';
import { type Event, LRUCache } from '@/deps.ts'; import { type Event, LRUCache } from '@/deps.ts';
import { publish } from '@/firehose.ts';
import { isEphemeralKind } from '@/kinds.ts'; import { isEphemeralKind } from '@/kinds.ts';
import * as mixer from '@/mixer.ts'; import * as mixer from '@/mixer.ts';
import { isLocallyFollowed } from '@/queries.ts'; import { isLocallyFollowed } from '@/queries.ts';
@ -27,6 +28,7 @@ async function handleEvent(event: Event): Promise<void> {
trackRelays(event), trackRelays(event),
trackHashtags(event), trackHashtags(event),
streamOut(event, data), streamOut(event, data),
broadcast(event, data),
]); ]);
} }
@ -130,6 +132,18 @@ function streamOut(event: Event, data: EventData) {
} }
} }
/**
* Publish the event to other relays.
* This should only be done in certain circumstances, like mentioning a user or publishing deletions.
*/
function broadcast(event: Event, data: EventData) {
if (!data.user || !isFresh(event)) return;
if (event.kind === 5) {
publish(event);
}
}
/** NIP-20 command line result. */ /** NIP-20 command line result. */
class RelayError extends Error { class RelayError extends Error {
constructor(prefix: 'duplicate' | 'pow' | 'blocked' | 'rate-limited' | 'invalid' | 'error', message: string) { constructor(prefix: 'duplicate' | 'pow' | 'blocked' | 'rate-limited' | 'invalid' | 'error', message: string) {