fix: cleanup before history sync

Author: michael-0acf4

src/typegate/src/main.ts

@@ -53,11 +53,6 @@ try {
     base: defaultTypegateConfigBase,
   });
   const typegate = await Typegate.init(config);
-  if (config.sync?.forceRemove) {
-    logger.warn("Force removal of typegraphs at boot");
-    await typegate.forceRemoveAllTypegraphs();
-  }
-
   await SystemTypegraph.loadAll(typegate, !globalConfig.packaged);
 
   const server = Deno.serve(

src/typegate/src/sync/replicated_map.ts

@@ -105,27 +105,43 @@ export class RedisReplicatedMap<T> implements AsyncDisposable {
     this.redisObs.close();
   }
 
+  async getAllHistory() {
+    const { key, redis } = this;
+    const all = await redis.hgetall(key);
+    const history = [];
+    for (let i = 0; i < all.length; i += 2) {
+      history.push({
+        name: all[i],
+        payload: all[i+1]
+      });
+    }
+
+    return history;
+  }
+
   async historySync(): Promise<XIdInput> {
-    const { key, redis, deserializer } = this;
+    const { redis, deserializer } = this;
 
     // get last received message before loading history
     const [lastMessage] = await redis.xrevrange(this.ekey, "+", "-", 1);
     const lastId = lastMessage ? lastMessage.xid : 0;
     logger.debug("last message loaded: {}", lastId);
 
-    const all = await redis.hgetall(key);
+    const all = await this.getAllHistory();
     logger.debug("history load start: {} elements", all.length);
-    for (let i = 0; i < all.length; i += 2) {
-      const name = all[i];
-      const payload = all[i + 1];
+
+    for (const { name, payload } of all) {
       logger.info(`reloaded addition: ${name}`);
       ensure(
         !this.memory.has(name),
         () => `typegraph ${name} should not exists in memory at first sync`,
       );
-      this.memory.set(name, await deserializer(payload, true));
+
+      const engine = await deserializer(payload, true);
+      this.memory.set(name, engine);
     }
     logger.debug("history load end");
+ 
     return lastId;
   }
 

src/typegate/src/typegate/mod.ts

@@ -43,7 +43,7 @@ import type { ArtifactStore } from "./artifacts/mod.ts";
 // TODO move from tests (MET-497)
 import { MemoryRegister } from "./memory_register.ts";
 import { NoLimiter } from "./no_limiter.ts";
-import { TypegraphStore } from "../sync/typegraph.ts";
+import { typegraphIdSchema, TypegraphStore } from "../sync/typegraph.ts";
 import { createLocalArtifactStore } from "./artifacts/local.ts";
 import { createSharedArtifactStore } from "./artifacts/shared.ts";
 import { AsyncDisposableStack } from "dispose";
@@ -141,15 +141,25 @@ export class Typegate implements AsyncDisposable {
         stack.move(),
       );
 
+      const typegraphStore = TypegraphStore.init(syncConfig, cryptoKeys);
       const register = await ReplicatedRegister.init(
         typegate,
         syncConfig.redis,
-        TypegraphStore.init(syncConfig, cryptoKeys),
+        typegraphStore
       );
       typegate.disposables.use(register);
 
       (typegate as { register: Register }).register = register;
 
+
+      if (config.sync?.forceRemove) {
+        logger.warn("Force removal at boot enabled");
+        const history = await register.replicatedMap.getAllHistory();
+        for (const { name, payload } of history) {
+          await typegate.forceRemove(name, payload, typegraphStore);
+        }
+      }
+
       const lastSync = await register.historySync().catch((err) => {
         logger.error(err);
         throw new Error(
@@ -397,10 +407,20 @@ export class Typegate implements AsyncDisposable {
     await this.artifactStore.runArtifactGC();
   }
 
-  async forceRemoveAllTypegraphs() {
-    for (const engine of this.register.list()) {
-      await this.removeTypegraph(engine.name);
-    }
+  async forceRemove(name: string, payload: string, typegraphStore: TypegraphStore) {
+    logger.warn(`Dropping "${name}": started`);
+    const typegraphId = typegraphIdSchema.parse(JSON.parse(payload));
+    const [tg] = await typegraphStore.download(
+      typegraphId,
+    );
+    const artifacts = new Set(
+      Object.values(tg.meta.artifacts).map((m) => m.hash),
+    );
+
+    await this.register.remove(name);
+    await this.artifactStore.updateRefCounts(new Set(), artifacts);
+    await this.artifactStore.runArtifactGC();
+    logger.warn(`Dropping "${name}": done`);
   }
 
   async initQueryEngine(

src/typegate/src/typegate/register.ts

@@ -86,7 +86,7 @@ export class ReplicatedRegister extends Register {
     return new ReplicatedRegister(replicatedMap);
   }
 
-  constructor(private replicatedMap: RedisReplicatedMap<QueryEngine>) {
+  constructor(public replicatedMap: RedisReplicatedMap<QueryEngine>) {
     super();
   }