Add a script for rebuilding the users table from scratch

Author: dogamak

package.json

@@ -33,7 +33,8 @@
     "run-docker": "docker run -p 8080:8080 --env-file=.env -d --name user-service-container user-service",
     "prepare": "[[ -z \"$NODE_END\" || \"$NODE_ENV\" = \"development\" ]] && husky || true",
     "knex": "tsx ./node_modules/knex/bin/cli.js",
-    "create-user": "tsx ./scripts/create_user.ts"
+    "create-user": "tsx ./scripts/create_user.ts",
+    "rebuild-users": "tsx ./scripts/rebuild_users.ts"
   },
   "dependencies": {
     "@sentry/node": "^8.25.0",

scripts/rebuild_users.ts

@@ -0,0 +1,8 @@
+import UserService from "../src/services/UserService";
+
+async function main() {
+  await UserService.rebuild();
+  process.exit(0);
+}
+
+main();

src/dao/UserDao.ts

@@ -241,6 +241,10 @@ export class UserDao implements Dao<UserDatabaseObject> {
     };
     return Promise.resolve(this.knex<UserDatabaseObject>(tableName).insert(savedObj));
   }
+
+  public async clear() {
+    await this.knex(tableName).del();
+  }
 }
 
 export default new UserDao();

src/services/NatsService.ts

@@ -103,6 +103,12 @@ export default class NatsService {
     await this.setup();
   }
 
+  async resetConsumer() {
+    const jsm = await this.conn.jetstreamManager();
+    await jsm.consumers.delete(STREAM, CONSUMER);
+    await this.setup();
+  }
+
   private async setup() {
     // Luodaan tarvittavat JetStream streamit ja consumerit.
 
@@ -298,14 +304,14 @@ export default class NatsService {
    * Tämän toteutus on jotenkin vaikeampi kuin NATS muuten antaisi olettaa,
    * joten luulen että tähän on joku fiksumpikin tapa.
    */
-  public async fetch(subject: string): Promise<JsMsg[]> {
+  public async fetch(subject: string, callback?: (msg: JsMsg) => Promise<void>): Promise<JsMsg[]> {
     const js = this.conn.jetstream();
 
     // Haetaan palvelimelta subjektin viimeisin viesti,
     // jotta saamme sen järjestysnumeron.
     const c1 = await js.consumers.get(STREAM, {
       // Haluamme vain subjektin viimeisimmän viestin.
-      deliver_policy: DeliverPolicy.LastPerSubject,
+      deliver_policy: DeliverPolicy.Last,
 
       // Haluamme vain yhden subjektin viestin.
       filterSubjects: [subject],
@@ -328,22 +334,18 @@ export default class NatsService {
 
     const results = [];
 
-    while (true) {
-      // Haetaan viesti kerrallaan.
-      const msg = await consumer.next();
-
-      // Ilmeisesti sieltä voi tulla myös null, en tiedä miksi.
-      if (!msg) break;
-
+    for await (const msg of await consumer.consume()) {
       // Jos se ei ollut null, niin sitten otetaan se talteen.
       results.push(msg);
 
+      await callback?.(msg);
+
       // Jos viestin järjestysnumero vastaa aiemmin hakemaamme viimeistintä viestiä,
       // lopetetaan viestien hakeminen.
       //
       // Tämän ei pitäisi olla tarpeellista, vaan NATS-ille pitäisi pystyä sanomaan myös,
       // että emme halua odottaa uusia viestejä. Enpäs saanut toimimaan, niin tehdään sitten näin.
-      if (msg.seq === last.seq) {
+      if (msg.seq === last.seq || msg.info.pending === 0) {
         break;
       }
     }

src/services/UserService.ts

@@ -383,6 +383,62 @@ class UserService {
     return this.dao.withTransaction(dao => callback(new UserService(dao)));
   }
 
+  public async rebuild() {
+    const nats = await NatsService.get();
+
+    await this.transaction(async tsx => {
+      await tsx.dao.clear();
+      await nats.fetch("members.*", async msg => {
+        const data = msg.json();
+        await tsx.handleMessage(data, msg);
+      });
+    });
+  }
+
+  public async handleMessage(pEvent: unknown, msg: JsMsg) {
+    const event = UserEvent.parse(pEvent);
+
+    const userId = parseInt(msg.subject.split(".")[1], 10);
+
+    if (event.type === "set") {
+      // Parsitaan käyttäjän ID viestin subjektista, joka on muotoa `members.{id}`.
+
+      if (event.fields.created) {
+        event.fields.created = new Date(event.fields.created);
+      }
+
+      if (event.fields.registration_ban_bypass_until) {
+        event.fields.registration_ban_bypass_until = new Date(event.fields.registration_ban_bypass_until);
+      }
+
+      // Päivitetään muokkausviestin mukaiset arvot tietokantaan.
+      await this.dao.update(userId, event.fields);
+    } else if (event.type === "create" || event.type === "import") {
+      if (event.fields.created) {
+        event.fields.created = new Date(event.fields.created);
+      }
+
+      if (event.fields.modified) {
+        event.fields.modified = new Date(event.fields.modified);
+      }
+
+      await this.dao.save({
+        ...event.fields,
+        id: userId,
+        last_seq: msg.seq,
+      });
+
+      return;
+    } else if (event.type === "delete") {
+      await this.dao.remove(userId);
+      return;
+    }
+
+    // Tallennetaan tietokantaan tieto siitä, että olemme käsitelleet tämän viestin,
+    // vaikka viesti ei olisikaan meille relevantti.
+    await this.dao.update(userId, { last_seq: msg.seq });
+  }
+
   /**
    * Taustaprosessi, joka kuuntelee jäsentieto-streamiin julkaistuja viestejä
    * ja käsittelee ne.
@@ -393,52 +449,7 @@ class UserService {
     this.abortSignal = new ConsumerAbortSignal();
 
     return new Promise<void>(resolve => {
-      const handler = async (pEvent: unknown, msg: JsMsg) => {
-        const event = UserEvent.parse(pEvent);
-
-        const userId = parseInt(msg.subject.split(".")[1], 10);
-
-        if (event.type === "set") {
-          // Parsitaan käyttäjän ID viestin subjektista, joka on muotoa `members.{id}`.
-
-          if (event.fields.created) {
-            // Tämä piti tehdä jostain syystä. Syyttäkää user-serviceä älkääkä NATSia.
-            event.fields.created = new Date(event.fields.created);
-          }
-
-          if (event.fields.registration_ban_bypass_until) {
-            event.fields.registration_ban_bypass_until = new Date(event.fields.registration_ban_bypass_until);
-          }
-
-          // Päivitetään muokkausviestin mukaiset arvot tietokantaan.
-          await this.dao.update(userId, event.fields);
-        } else if (event.type === "create" || event.type === "import") {
-          if (event.fields.created) {
-            event.fields.created = new Date(event.fields.created);
-          }
-
-          if (event.fields.modified) {
-            event.fields.modified = new Date(event.fields.modified);
-          }
-
-          await this.dao.save({
-            ...event.fields,
-            id: userId,
-            last_seq: msg.seq,
-          });
-
-          return;
-        } else if (event.type === "delete") {
-          await this.dao.remove(userId);
-          return;
-        }
-
-        // Tallennetaan tietokantaan tieto siitä, että olemme käsitelleet tämän viestin,
-        // vaikka viesti ei olisikaan meille relevantti.
-        await this.dao.update(userId, { last_seq: msg.seq });
-      };
-
-      nats.subscribe(handler, {
+      nats.subscribe(this.handleMessage.bind(this), {
         onReady: () => resolve(),
         signal: this.abortSignal,
       });