Merge branch 'rm-mixer' into 'main'

Remove mixer module

See merge request soapbox-pub/ditto!87

src/controllers/api/accounts.ts

@@ -1,9 +1,9 @@
 import { type AppController } from '@/app.ts';
 import { Conf } from '@/config.ts';
+import * as eventsDB from '@/db/events.ts';
 import { insertUser } from '@/db/users.ts';
 import { findReplyTag, nip19, z } from '@/deps.ts';
 import { type DittoFilter } from '@/filter.ts';
-import * as mixer from '@/mixer.ts';
 import { getAuthor, getFollowedPubkeys, getFollows } from '@/queries.ts';
 import { booleanParamSchema, fileSchema } from '@/schema.ts';
 import { jsonMetaContentSchema } from '@/schemas/nostr.ts';
@@ -151,7 +151,7 @@ const accountStatusesController: AppController = async (c) => {
     filter['#t'] = [tagged];
   }
 
-  let events = await mixer.getFilters([filter]);
+  let events = await eventsDB.getFilters([filter]);
 
   if (exclude_replies) {
     events = events.filter((event) => !findReplyTag(event));
@@ -256,7 +256,7 @@ const favouritesController: AppController = async (c) => {
   const pubkey = c.get('pubkey')!;
   const params = paginationSchema.parse(c.req.query());
 
-  const events7 = await mixer.getFilters(
+  const events7 = await eventsDB.getFilters(
     [{ kinds: [7], authors: [pubkey], ...params }],
     { signal: AbortSignal.timeout(1000) },
   );
@@ -265,9 +265,12 @@ const favouritesController: AppController = async (c) => {
     .map((event) => event.tags.find((tag) => tag[0] === 'e')?.[1])
     .filter((id): id is string => !!id);
 
-  const events1 = await mixer.getFilters([{ kinds: [1], ids, relations: ['author', 'event_stats', 'author_stats'] }], {
-    signal: AbortSignal.timeout(1000),
-  });
+  const events1 = await eventsDB.getFilters(
+    [{ kinds: [1], ids, relations: ['author', 'event_stats', 'author_stats'] }],
+    {
+      signal: AbortSignal.timeout(1000),
+    },
+  );
 
   const statuses = await Promise.all(events1.map((event) => renderStatus(event, c.get('pubkey'))));
   return paginated(c, events1, statuses);

src/controllers/api/notifications.ts

@@ -1,5 +1,5 @@
 import { type AppController } from '@/app.ts';
-import * as mixer from '@/mixer.ts';
+import * as eventsDB from '@/db/events.ts';
 import { paginated, paginationSchema } from '@/utils/web.ts';
 import { renderNotification } from '@/views/mastodon/notifications.ts';
 
@@ -7,7 +7,7 @@ const notificationsController: AppController = async (c) => {
   const pubkey = c.get('pubkey')!;
   const { since, until } = paginationSchema.parse(c.req.query());
 
-  const events = await mixer.getFilters(
+  const events = await eventsDB.getFilters(
     [{ kinds: [1], '#p': [pubkey], since, until }],
     { signal: AbortSignal.timeout(3000) },
   );

src/controllers/api/search.ts

@@ -2,7 +2,6 @@ import { AppController } from '@/app.ts';
 import * as eventsDB from '@/db/events.ts';
 import { type Event, nip19, z } from '@/deps.ts';
 import { type DittoFilter } from '@/filter.ts';
-import * as mixer from '@/mixer.ts';
 import { booleanParamSchema } from '@/schema.ts';
 import { nostrIdSchema } from '@/schemas/nostr.ts';
 import { dedupeEvents } from '@/utils.ts';
@@ -95,7 +94,7 @@ function typeToKinds(type: SearchQuery['type']): number[] {
 /** Resolve a searched value into an event, if applicable. */
 async function lookupEvent(query: SearchQuery, signal = AbortSignal.timeout(1000)): Promise<Event | undefined> {
   const filters = await getLookupFilters(query);
-  const [event] = await mixer.getFilters(filters, { limit: 1, signal });
+  const [event] = await eventsDB.getFilters(filters, { limit: 1, signal });
   return event;
 }
 

src/controllers/api/timelines.ts

@@ -1,6 +1,6 @@
+import * as eventsDB from '@/db/events.ts';
 import { z } from '@/deps.ts';
 import { type DittoFilter } from '@/filter.ts';
-import * as mixer from '@/mixer.ts';
 import { getFeedPubkeys } from '@/queries.ts';
 import { booleanParamSchema } from '@/schema.ts';
 import { paginated, paginationSchema } from '@/utils/web.ts';
@@ -33,7 +33,7 @@ const hashtagTimelineController: AppController = (c) => {
 
 /** Render statuses for timelines. */
 async function renderStatuses(c: AppContext, filters: DittoFilter<1>[], signal = AbortSignal.timeout(1000)) {
-  const events = await mixer.getFilters(
+  const events = await eventsDB.getFilters(
     filters.map((filter) => ({ ...filter, relations: ['author', 'event_stats', 'author_stats'] })),
     { signal },
   );

src/mixer.ts

@@ -1,61 +0,0 @@
-import { type Event, matchFilters } from '@/deps.ts';
-
-import * as eventsDB from '@/db/events.ts';
-import { dedupeEvents, eventDateComparator } from '@/utils.ts';
-
-import type { DittoFilter, GetFiltersOpts } from '@/filter.ts';
-
-/** Get filters from the database and pool, and mix the best results together. */
-async function getFilters<K extends number>(
-  filters: DittoFilter<K>[],
-  opts?: GetFiltersOpts,
-): Promise<Event<K>[]> {
-  if (!filters.length) return Promise.resolve([]);
-
-  const results = await Promise.allSettled([
-    eventsDB.getFilters(filters, opts),
-  ]);
-
-  const events = results
-    .filter((result): result is PromiseFulfilledResult<Event<K>[]> => result.status === 'fulfilled')
-    .flatMap((result) => result.value);
-
-  return unmixEvents(events, filters);
-}
-
-/** Combine and sort events to match the filters. */
-function unmixEvents<K extends number>(events: Event<K>[], filters: DittoFilter<K>[]): Event<K>[] {
-  events = dedupeEvents(events);
-  events = takeNewestEvents(events);
-  events = events.filter((event) => matchFilters(filters, event));
-  events.sort(eventDateComparator);
-
-  return events;
-}
-
-/** Take the newest events among replaceable ones. */
-function takeNewestEvents<K extends number>(events: Event<K>[]): Event<K>[] {
-  const isReplaceable = (kind: number) =>
-    kind === 0 || kind === 3 || (10000 <= kind && kind < 20000) || (30000 <= kind && kind < 40000);
-
-  // Group events by author and kind.
-  const groupedEvents = events.reduce<Map<string, Event<K>[]>>((acc, event) => {
-    const key = `${event.pubkey}:${event.kind}`;
-    const group = acc.get(key) || [];
-    acc.set(key, [...group, event]);
-    return acc;
-  }, new Map());
-
-  // Process each group.
-  const processedEvents = Array.from(groupedEvents.values()).flatMap((group) => {
-    if (isReplaceable(group[0].kind)) {
-      // Sort by `created_at` and take the latest event.
-      return group.sort(eventDateComparator)[0];
-    }
-    return group;
-  });
-
-  return processedEvents;
-}
-
-export { getFilters };

src/pipeline.ts

@@ -6,7 +6,6 @@ import { deleteAttachedMedia } from '@/db/unattached-media.ts';
 import { findUser } from '@/db/users.ts';
 import { Debug, type Event } from '@/deps.ts';
 import { isEphemeralKind } from '@/kinds.ts';
-import * as mixer from '@/mixer.ts';
 import { publish } from '@/pool.ts';
 import { isLocallyFollowed } from '@/queries.ts';
 import { reqmeister } from '@/reqmeister.ts';
@@ -70,7 +69,7 @@ async function storeEvent(event: Event, data: EventData, opts: StoreEventOpts =
   const { force = false } = opts;
 
   if (force || data.user || isAdminEvent(event) || await isLocallyFollowed(event.pubkey)) {
-    const [deletion] = await mixer.getFilters(
+    const [deletion] = await eventsDB.getFilters(
       [{ kinds: [5], authors: [event.pubkey], '#e': [event.id], limit: 1 }],
       { limit: 1, signal: AbortSignal.timeout(Time.seconds(1)) },
     );

src/queries.ts

@@ -1,7 +1,6 @@
 import * as eventsDB from '@/db/events.ts';
 import { type Event, findReplyTag } from '@/deps.ts';
 import { type AuthorMicrofilter, type DittoFilter, type IdMicrofilter, type Relation } from '@/filter.ts';
-import * as mixer from '@/mixer.ts';
 import { reqmeister } from '@/reqmeister.ts';
 import { memorelay } from '@/db/memorelay.ts';
 
@@ -79,7 +78,7 @@ const getAuthor = async (pubkey: string, opts: GetEventOpts<0> = {}): Promise<Ev
 
 /** Get users the given pubkey follows. */
 const getFollows = async (pubkey: string, signal = AbortSignal.timeout(1000)): Promise<Event<3> | undefined> => {
-  const [event] = await mixer.getFilters([{ authors: [pubkey], kinds: [3], limit: 1 }], { limit: 1, signal });
+  const [event] = await eventsDB.getFilters([{ authors: [pubkey], kinds: [3], limit: 1 }], { limit: 1, signal });
   return event;
 };
 
@@ -118,7 +117,7 @@ async function getAncestors(event: Event<1>, result = [] as Event<1>[]): Promise
 }
 
 function getDescendants(eventId: string, signal = AbortSignal.timeout(2000)): Promise<Event<1>[]> {
-  return mixer.getFilters(
+  return eventsDB.getFilters(
     [{ kinds: [1], '#e': [eventId], relations: ['author', 'event_stats', 'author_stats'] }],
     { limit: 200, signal },
   );

src/views.ts

@@ -1,13 +1,13 @@
 import { AppContext } from '@/app.ts';
+import * as eventsDB from '@/db/events.ts';
 import { type Filter } from '@/deps.ts';
-import * as mixer from '@/mixer.ts';
 import { getAuthor } from '@/queries.ts';
 import { renderAccount } from '@/views/mastodon/accounts.ts';
 import { paginated } from '@/utils/web.ts';
 
 /** Render account objects for the author of each event. */
 async function renderEventAccounts(c: AppContext, filters: Filter[]) {
-  const events = await mixer.getFilters(filters);
+  const events = await eventsDB.getFilters(filters);
   const pubkeys = new Set(events.map(({ pubkey }) => pubkey));
 
   if (!pubkeys.size) {