Discord: fetch backwards and add a beginningDate config

packages/sourcecred/src/plugins/discord/config.js

@@ -9,6 +9,12 @@ import {
   type ChannelWeightConfig,
   type WeightConfig,
 } from "./reactionWeights";
+import {
+  fromISO,
+  timestampISOParser,
+  type TimestampISO,
+  type TimestampMs,
+} from "../../util/timestamp";
 
 export type {BotToken as DiscordToken} from "./models";
 
@@ -71,6 +77,8 @@ export type DiscordConfigJson = $ReadOnlyArray<{|
   +includeNsfwChannels?: boolean,
   // This reduces graph size by replacing reaction nodes with message node weights.
   +simplifyGraph?: boolean,
+  // This tells the fetcher not to fetch any messages before this date.
+  +beginningDate?: TimestampISO,
 |}>;
 
 const parserJson: C.Parser<DiscordConfigJson> = C.array(
@@ -100,6 +108,7 @@ const parserJson: C.Parser<DiscordConfigJson> = C.array(
       propsChannels: C.array(C.string),
       includeNsfwChannels: C.boolean,
       simplifyGraph: C.boolean,
+      beginningDate: timestampISOParser,
     }
   )
 );
@@ -110,6 +119,7 @@ export type DiscordConfig = {|
   +propsChannels: $ReadOnlyArray<Model.Snowflake>,
   +includeNsfwChannels: boolean,
   +simplifyGraph: boolean,
+  +beginningTimestampMs: TimestampMs,
 |};
 export type DiscordConfigs = $ReadOnlyArray<DiscordConfig>;
 
@@ -144,6 +154,9 @@ export function _upgrade(json: DiscordConfigJson): DiscordConfigs {
     propsChannels: config.propsChannels || [],
     includeNsfwChannels: config.includeNsfwChannels || false,
     simplifyGraph: config.simplifyGraph || false,
+    beginningTimestampMs: config.beginningDate
+      ? fromISO(config.beginningDate)
+      : -Infinity,
   }));
 }
 

packages/sourcecred/src/plugins/discord/fetcher.js

@@ -11,7 +11,7 @@ export interface DiscordApi {
   members(guild: Model.Snowflake): Promise<$ReadOnlyArray<Model.GuildMember>>;
   messages(
     channel: Model.Snowflake,
-    after: Model.Snowflake,
+    before: Model.Snowflake,
     limit: number
   ): Promise<$ReadOnlyArray<Model.Message>>;
   reactions(
@@ -72,7 +72,7 @@ export class Fetcher implements DiscordApi {
           waitTime / 1000
         } seconds (until ${restartDate.toLocaleString()})`
       );
-      await new Promise((resolve) => setTimeout(resolve, waitTime));
+      await new Promise((resolve) => setTimeout(resolve, waitTime + 1));
     }
   }
 
@@ -170,12 +170,14 @@ export class Fetcher implements DiscordApi {
 
   async messages(
     channel: Model.Snowflake,
-    after: Model.Snowflake,
+    beforeId: Model.Snowflake,
     limit: number
   ): Promise<$ReadOnlyArray<Model.Message>> {
     const messages = await this._fetchJson(
-      `/channels/${channel}/messages?after=${after}&limit=${limit}`
+      `/channels/${channel}/messages?limit=${limit}` +
+        (beforeId ? `&before=${beforeId}` : "")
     );
+
     return messages.map((x) => ({
       id: x.id,
       channelId: channel,

packages/sourcecred/src/plugins/discord/mirror.js

@@ -9,6 +9,8 @@ import {channelWeight} from "./reactionWeights";
 
 // How many messages for each channel to reload.
 const RELOAD_DEPTH = 50;
+// How many messages to fetch at a time.
+const MESSAGE_FETCH_LIMIT = 100;
 
 export class Mirror {
   +_repo: SqliteMirrorRepository;
@@ -33,7 +35,7 @@ export class Mirror {
     for (const channel of channels) {
       reporter.start(`discord/${guild.name}/#${channel.name}`);
       try {
-        await this.addMessages(channel.id, channel.parentId);
+        await this.addMessages(channel.id);
       } catch (e) {
         const warn = e?.message?.includes("403")
           ? "Skipping private channel."
@@ -88,33 +90,51 @@ export class Mirror {
   }
 
   async addMessages(
-    channel: Model.Snowflake,
-    category?: Model.Snowflake,
-    messageLimit?: number
+    channel: Model.Snowflake
   ): Promise<$ReadOnlyArray<Model.Message>> {
-    const loadStart = this._repo.nthMessageFromTail(channel, RELOAD_DEPTH);
-    // console.log(channel, (loadStart || {}).id);
-
-    const limit = messageLimit || 100;
-    let page: $ReadOnlyArray<Model.Message> = [];
-    let after: Model.Snowflake = loadStart ? loadStart.id : "0";
-    do {
-      page = await this._api.messages(channel, after, limit);
-      for (const message of page) {
-        after = after < message.id ? message.id : after;
-        this._repo.addMessage(message);
-        for (const emoji of message.reactionEmoji) {
-          const reactions = await this._api.reactions(
-            channel,
-            message.id,
-            emoji
-          );
-          for (const reaction of reactions) {
-            this._repo.addReaction(reaction);
+    let beginningTimestampMs =
+      this._repo.nthMessageFromTail(channel, RELOAD_DEPTH)?.timestampMs ??
+      this.config.beginningTimestampMs;
+    let beforeId: Model.Snowflake = "";
+    const run = async (_this) => {
+      let page: $ReadOnlyArray<Model.Message> = [];
+      let hasReachedBeginning = false;
+      do {
+        page = await _this._api.messages(
+          channel,
+          beforeId,
+          MESSAGE_FETCH_LIMIT
+        );
+        for (const message of page) {
+          if (message.timestampMs < beginningTimestampMs) {
+            hasReachedBeginning = true;
+            continue;
+          }
+          if (!beforeId || beforeId > message.id) {
+            beforeId = message.id;
+          }
+          _this._repo.addMessage(message);
+          for (const emoji of message.reactionEmoji) {
+            const reactions = await _this._api.reactions(
+              channel,
+              message.id,
+              emoji
+            );
+            for (const reaction of reactions) {
+              _this._repo.addReaction(reaction);
+            }
           }
         }
-      }
-    } while (page.length >= limit);
+      } while (page.length >= MESSAGE_FETCH_LIMIT && !hasReachedBeginning);
+    };
+    await run(this);
+
+    const firstMessage = this._repo.nthMessageFromTail(channel, Infinity);
+    if (firstMessage) {
+      beforeId = firstMessage.id;
+      beginningTimestampMs = this.config.beginningTimestampMs;
+      await run(this);
+    }
     return this._repo.messages(channel);
   }
 }

packages/sourcecred/src/plugins/discord/mirror.test.js

@@ -29,6 +29,7 @@ describe("plugins/discord/mirror", () => {
     },
     includeNsfwChannels: true,
     simplifyGraph: true,
+    beginningTimestampMs: -Infinity,
   };
 
   describe("smoke test", () => {

packages/sourcecred/src/plugins/discord/mirrorRepository.js

@@ -224,8 +224,8 @@ export class SqliteMirrorRepository {
       )
       .get({channel_id: channel}).count;
 
-    if (count < n) return null;
-    const offset = count - n;
+    if (count === 0) return null;
+    const offset = count < n ? 0 : count - n;
     const m = this._db
       .prepare(
         dedent`\