Merge branch 'trending-tags' into 'develop'

Add trending hashtags

See merge request soapbox-pub/ditto!2

.gitlab-ci.yml

@@ -14,6 +14,6 @@ lint:
   stage: test
   script: deno lint
 
-# test:
+test:
-#   stage: test
+  stage: test
-#   script: deno task test
+  script: deno task test

data/.gitignore

@@ -0,0 +1,2 @@
+*
+!.gitignore

deno.json

@@ -2,8 +2,8 @@
   "$schema": "https://deno.land/x/deno@v1.32.3/cli/schemas/config-file.v1.json",
   "lock": false,
   "tasks": {
-    "dev": "deno run --allow-read --allow-env --allow-net --allow-ffi --unstable --watch src/server.ts",
+    "dev": "deno run --allow-read --allow-write --allow-env --allow-net --allow-ffi --unstable --watch src/server.ts",
-    "test": "deno test"
+    "test": "deno test -A --unstable src"
   },
   "imports": {
     "@/": "./src/"

src/app.ts

@@ -1,5 +1,6 @@
 import { type Context, cors, type Handler, Hono, type HonoEnv, logger, type MiddlewareHandler } from '@/deps.ts';
 import { type Event } from '@/event.ts';
+import '@/loopback.ts';
 
 import {
   accountController,
@@ -26,6 +27,7 @@ import {
   statusController,
 } from './controllers/api/statuses.ts';
 import { streamingController } from './controllers/api/streaming.ts';
+import { trendingTagsController } from './controllers/api/trends.ts';
 import { indexController } from './controllers/site.ts';
 import { hostMetaController } from './controllers/well-known/host-meta.ts';
 import { nodeInfoController, nodeInfoSchemaController } from './controllers/well-known/nodeinfo.ts';
@@ -100,6 +102,9 @@ app.get('/api/v2/search', searchController);
 
 app.get('/api/pleroma/frontend_configurations', frontendConfigController);
 
+app.get('/api/v1/trends/tags', trendingTagsController);
+app.get('/api/v1/trends', trendingTagsController);
+
 // Not (yet) implemented.
 app.get('/api/v1/notifications', emptyArrayController);
 app.get('/api/v1/bookmarks', emptyArrayController);

src/client.ts

@@ -172,7 +172,7 @@ function getDescendants(eventId: string): Promise<SignedEvent<1>[]> {
 
 /** Publish an event to the Nostr relay. */
 function publish(event: SignedEvent, relays = Conf.publishRelays): void {
-  console.log('Publishing event', event);
+  console.log('Publishing event', event, relays);
   try {
     getPool().publish(event, relays);
   } catch (e) {

src/config.ts

@@ -4,7 +4,11 @@ const Conf = {
     return Deno.env.get('DITTO_NSEC');
   },
   get relay() {
-    return Deno.env.get('DITTO_RELAY');
+    const value = Deno.env.get('DITTO_RELAY');
+    if (!value) {
+      throw new Error('Missing DITTO_RELAY');
+    }
+    return value;
   },
   get localDomain() {
     return Deno.env.get('LOCAL_DOMAIN') || 'http://localhost:8000';

src/controllers/api/trends.ts

@@ -0,0 +1,47 @@
+import { type AppController } from '@/app.ts';
+import { Conf } from '@/config.ts';
+import { z } from '@/deps.ts';
+import { trends } from '@/trends.ts';
+import { Time } from '@/utils.ts';
+
+const limitSchema = z.coerce.number().catch(10).transform((value) => Math.min(Math.max(value, 0), 20));
+
+const trendingTagsController: AppController = (c) => {
+  const limit = limitSchema.parse(c.req.query('limit'));
+  if (limit < 1) return c.json([]);
+
+  const now = new Date();
+  const yesterday = new Date(now.getTime() - Time.days(1));
+  const lastWeek = new Date(now.getTime() - Time.days(7));
+
+  const tags = trends.getTrendingTags({
+    since: yesterday,
+    until: now,
+    limit,
+  });
+
+  return c.json(tags.map(({ name, uses, accounts }) => ({
+    name,
+    url: Conf.local(`/tags/${name}`),
+    history: [
+      {
+        day: String(Math.floor(now.getTime() / 1000)),
+        accounts: String(accounts),
+        uses: String(uses),
+      },
+      ...trends.getTagHistory({
+        tag: name,
+        since: lastWeek,
+        until: now,
+        limit: 6,
+        offset: 1,
+      }).map((history) => ({
+        day: String(Math.floor(history.day.getTime() / 1000)),
+        accounts: String(history.accounts),
+        uses: String(history.uses),
+      })),
+    ],
+  })));
+};
+
+export { trendingTagsController };

src/db.ts

@@ -1,9 +1,10 @@
 import { createPentagon, z } from '@/deps.ts';
+import { hexIdSchema } from '@/schema.ts';
 
 const kv = await Deno.openKv();
 
 const userSchema = z.object({
-  pubkey: z.string().regex(/^[0-9a-f]{64}$/).describe('primary'),
+  pubkey: hexIdSchema.describe('primary'),
   username: z.string().regex(/^\w{1,30}$/).describe('unique'),
   createdAt: z.date(),
 });

src/deps-test.ts

@@ -0,0 +1 @@
+export { assert, assertEquals, assertThrows } from 'https://deno.land/std@0.177.0/testing/asserts.ts';

src/deps.ts

@@ -39,3 +39,5 @@ export { default as sanitizeHtml } from 'npm:sanitize-html@^2.10.0';
 export { default as ISO6391 } from 'npm:iso-639-1@2.1.15';
 export { Dongoose } from 'https://raw.githubusercontent.com/alexgleason/dongoose/68b7ad9dd7b6ec0615e246a9f1603123c1709793/mod.ts';
 export { createPentagon } from 'https://deno.land/x/pentagon@v0.1.1/mod.ts';
+export { DB as Sqlite } from 'https://deno.land/x/sqlite@v3.7.0/mod.ts';
+export { serve } from 'https://deno.land/std@0.177.0/http/server.ts';

src/loopback.ts

@@ -0,0 +1,44 @@
+import { Conf } from '@/config.ts';
+import { RelayPool } from '@/deps.ts';
+import { trends } from '@/trends.ts';
+import { nostrDate, nostrNow } from '@/utils.ts';
+
+import type { Event } from '@/event.ts';
+
+const relay = new RelayPool([Conf.relay]);
+
+// This file watches all events on your Ditto relay and triggers
+// side-effects based on them. This can be used for things like
+// notifications, trending hashtag tracking, etc.
+relay.subscribe(
+  [{ kinds: [1], since: nostrNow() }],
+  [Conf.relay],
+  handleEvent,
+  undefined,
+  undefined,
+);
+
+/** Handle events through the loopback pipeline. */
+function handleEvent(event: Event): void {
+  console.info('loopback event:', event.id);
+  trackHashtags(event);
+}
+
+/** Track whenever a hashtag is used, for processing trending tags. */
+function trackHashtags(event: Event): void {
+  const date = nostrDate(event.created_at);
+
+  const tags = event.tags
+    .filter((tag) => tag[0] === 't')
+    .map((tag) => tag[1])
+    .slice(0, 5);
+
+  if (!tags.length) return;
+
+  try {
+    console.info('tracking tags:', tags);
+    trends.addTagUsages(event.pubkey, tags, date);
+  } catch (_e) {
+    // do nothing
+  }
+}

src/schema.ts

@@ -67,15 +67,15 @@ const relaySchema = z.custom<URL>((relay) => {
   }
 });
 
-const nostrIdSchema = z.string().regex(/^[0-9a-f]{64}$/);
+const hexIdSchema = z.string().regex(/^[0-9a-f]{64}$/);
 
 const eventSchema = z.object({
-  id: nostrIdSchema,
+  id: hexIdSchema,
   kind: z.number(),
   tags: z.array(z.array(z.string())),
   content: z.string(),
   created_at: z.number(),
-  pubkey: nostrIdSchema,
+  pubkey: hexIdSchema,
   sig: z.string(),
 });
 
@@ -95,10 +95,14 @@ const decode64Schema = z.string().transform((value, ctx) => {
   }
 });
 
+const hashtagSchema = z.string().regex(/^\w{1,30}$/);
+
 export {
   decode64Schema,
   emojiTagSchema,
   filteredArray,
+  hashtagSchema,
+  hexIdSchema,
   jsonSchema,
   type MetaContent,
   metaContentSchema,

src/server.ts

@@ -1,5 +1,5 @@
 import 'https://deno.land/std@0.177.0/dotenv/load.ts';
-import { serve } from 'https://deno.land/std@0.177.0/http/server.ts';
+import { serve } from '@/deps.ts';
 
 import app from './app.ts';
 

src/trends.test.ts

@@ -0,0 +1,32 @@
+import { assertEquals } from '@/deps-test.ts';
+import { Sqlite } from '@/deps.ts';
+
+import { TrendsDB } from './trends.ts';
+
+const db = new Sqlite(':memory:');
+const trends = new TrendsDB(db);
+
+const p8 = (pubkey8: string) => `${pubkey8}00000000000000000000000000000000000000000000000000000000`;
+
+Deno.test('getTrendingTags', () => {
+  trends.addTagUsages(p8('00000000'), ['ditto', 'hello', 'yolo']);
+  trends.addTagUsages(p8('00000000'), ['hello']);
+  trends.addTagUsages(p8('00000001'), ['Ditto', 'hello']);
+  trends.addTagUsages(p8('00000010'), ['DITTO']);
+
+  const result = trends.getTrendingTags({
+    since: new Date('1999-01-01T00:00:00'),
+    until: new Date('2999-01-01T00:00:00'),
+    threshold: 1,
+  });
+
+  const expected = [
+    { name: 'ditto', accounts: 3, uses: 3 },
+    { name: 'hello', accounts: 2, uses: 3 },
+    { name: 'yolo', accounts: 1, uses: 1 },
+  ];
+
+  assertEquals(result, expected);
+
+  trends.cleanupTagUsages(new Date('2999-01-01T00:00:00'));
+});

src/trends.ts

@@ -0,0 +1,106 @@
+import { Sqlite } from '@/deps.ts';
+import { hashtagSchema, hexIdSchema } from '@/schema.ts';
+import { Time } from './utils.ts';
+
+interface GetTrendingTagsOpts {
+  since: Date;
+  until: Date;
+  limit?: number;
+  threshold?: number;
+}
+
+interface GetTagHistoryOpts {
+  tag: string;
+  since: Date;
+  until: Date;
+  limit?: number;
+  offset?: number;
+}
+
+class TrendsDB {
+  #db: Sqlite;
+
+  constructor(db: Sqlite) {
+    this.#db = db;
+
+    this.#db.execute(`
+      CREATE TABLE IF NOT EXISTS tag_usages (
+        tag TEXT NOT NULL COLLATE NOCASE,
+        pubkey8 TEXT NOT NULL,
+        inserted_at DATETIME NOT NULL
+      );
+
+      CREATE INDEX IF NOT EXISTS idx_time_tag ON tag_usages(inserted_at, tag);
+    `);
+
+    const cleanup = () => {
+      console.info('Cleaning up old tag usages...');
+      const lastWeek = new Date(new Date().getTime() - Time.days(7));
+      this.cleanupTagUsages(lastWeek);
+    };
+
+    setInterval(cleanup, Time.hours(1));
+    cleanup();
+  }
+
+  getTrendingTags({ since, until, limit = 10, threshold = 3 }: GetTrendingTagsOpts) {
+    return this.#db.query<string[]>(
+      `
+      SELECT tag, COUNT(DISTINCT pubkey8), COUNT(*)
+        FROM tag_usages
+        WHERE inserted_at >= ? AND inserted_at < ?
+        GROUP BY tag
+        HAVING COUNT(DISTINCT pubkey8) >= ?
+        ORDER BY COUNT(DISTINCT pubkey8)
+        DESC LIMIT ?;
+    `,
+      [since, until, threshold, limit],
+    ).map((row) => ({
+      name: row[0],
+      accounts: Number(row[1]),
+      uses: Number(row[2]),
+    }));
+  }
+
+  getTagHistory({ tag, since, until, limit = 7, offset = 0 }: GetTagHistoryOpts) {
+    return this.#db.query<string[]>(
+      `
+      SELECT inserted_at, COUNT(DISTINCT pubkey8), COUNT(*)
+        FROM tag_usages
+        WHERE tag = ? AND inserted_at >= ? AND inserted_at < ?
+        GROUP BY date(inserted_at)
+        ORDER BY date(inserted_at) DESC
+        LIMIT ?
+        OFFSET ?;
+    `,
+      [tag, since, until, limit, offset],
+    ).map((row) => ({
+      day: new Date(row[0]),
+      accounts: Number(row[1]),
+      uses: Number(row[2]),
+    }));
+  }
+
+  addTagUsages(pubkey: string, hashtags: string[], date = new Date()): void {
+    const pubkey8 = hexIdSchema.parse(pubkey).substring(0, 8);
+    const tags = hashtagSchema.array().min(1).parse(hashtags);
+
+    this.#db.query(
+      'INSERT INTO tag_usages (tag, pubkey8, inserted_at) VALUES ' + tags.map(() => '(?, ?, ?)').join(', '),
+      tags.map((tag) => [tag, pubkey8, date]).flat(),
+    );
+  }
+
+  cleanupTagUsages(until: Date): void {
+    this.#db.query(
+      'DELETE FROM tag_usages WHERE inserted_at < ?',
+      [until],
+    );
+  }
+}
+
+const trends = new TrendsDB(
+  new Sqlite('data/trends.sqlite3'),
+);
+
+export { trends, TrendsDB };

src/utils.ts

@@ -81,7 +81,7 @@ async function parseBody(req: Request): Promise<unknown> {
 const paginationSchema = z.object({
   since: z.coerce.number().optional().catch(undefined),
   until: z.lazy(() => z.coerce.number().catch(nostrNow())),
-  limit: z.coerce.number().min(0).max(40).catch(20),
+  limit: z.coerce.number().catch(20).transform((value) => Math.min(Math.max(value, 0), 40)),
 });
 
 type PaginationParams = z.infer<typeof paginationSchema>;