fix: archive ownerless repositories without assigning ownership (#800)

src/core/GitHubUtils.ts

@@ -274,6 +274,7 @@ export async function checkToken(token: string) {
 const checkedRepositoryTokens = new WeakMap<Repository, string>();
 
 export async function getToken(repository: Repository) {
+  repository.assertNotArchived();
   logger.debug("getToken", { repoId: repository.repoId });
   const credential = await getCredential(repository.owner.id);
   const ownerAccessToken = credential?.token;

src/core/Repository.ts

@@ -71,10 +71,12 @@ export default class Repository {
   }
 
   async getToken() {
+    this.assertNotArchived();
     return getToken(this);
   }
 
   get source() {
+    this.assertNotArchived();
     const ghRepo = new GitHubRepository({
       name: this.model.source.repositoryName,
     });
@@ -130,6 +132,7 @@ export default class Repository {
       force: false,
     }
   ): Promise<IFile[]> {
+    this.assertNotArchived();
     const terms = this._model.options.terms || [];
     let hasFile = await FileModel.exists({ repoId: this.repoId }).exec();
     // Files created by GitHubDownload don't carry a valid 40-char GitHub
@@ -208,7 +211,14 @@ export default class Repository {
   /**
    * Check the status of the repository
    */
+  assertNotArchived() {
+    if (this.status === RepositoryStatus.ARCHIVED) {
+      throw new AnonymousError("repository_archived", { httpStatus: 410 });
+    }
+  }
+
   async check() {
+    this.assertNotArchived();
     if (
       this._model.options.expirationMode !== "never" &&
       this.status == RepositoryStatus.READY &&
@@ -258,6 +268,7 @@ export default class Repository {
    * @returns A stream of anonymized repository compressed
    */
   zip(): Promise<Readable> {
+    this.assertNotArchived();
     return storage.archive(this.repoId, "", {
       format: "zip",
       fileTransformer: (filename: string) =>
@@ -293,6 +304,7 @@ export default class Repository {
    * @returns void
    */
   async updateIfNeeded(opt?: { force: boolean }): Promise<void> {
+    this.assertNotArchived();
     if (
       this._model.options.expirationMode !== "never" &&
       this.status != RepositoryStatus.EXPIRED &&
@@ -415,6 +427,7 @@ export default class Repository {
    * @returns void
    */
   async anonymize(progress?: (status: string) => void) {
+    this.assertNotArchived();
     if (this.status === RepositoryStatus.READY) {
       return;
     }
@@ -459,6 +472,7 @@ export default class Repository {
   public protectLifecycle = false;
 
   async updateStatus(status: RepositoryStatus, statusMessage?: string) {
+    if (status !== RepositoryStatus.ARCHIVED) this.assertNotArchived();
     if (!status) return this.model;
     const statusDate = new Date();
     if (isConnected) {
@@ -466,7 +480,7 @@ export default class Repository {
         {
           _id: this._model._id,
           ...(this.protectLifecycle ? {
-            status: { $nin: [RepositoryStatus.REMOVING, RepositoryStatus.REMOVED,
+            status: { $nin: [RepositoryStatus.ARCHIVED, RepositoryStatus.REMOVING, RepositoryStatus.REMOVED,
               RepositoryStatus.EXPIRING, RepositoryStatus.EXPIRED] },
             anonymizeDate: this._model.anonymizeDate,
           } : {}),
@@ -504,6 +518,7 @@ export default class Repository {
    * Reset/delete the state of the repository
    */
   async resetSate(status?: RepositoryStatus, statusMessage?: string) {
+    this.assertNotArchived();
     // remove attribute
     this._model.size = { storage: 0, file: 0 };
     if (status) {
@@ -522,6 +537,8 @@ export default class Repository {
    * @returns
    */
   async removeCache() {
+    // Archive cleanup is handled by the resumable recovery script. Preserve DB metadata.
+    if (this.status === RepositoryStatus.ARCHIVED) return;
     await storage.rm(this.repoId);
     this.model.isReseted = true;
     this.model.size = { storage: 0, file: 0 };

src/core/migrate-credentials.ts

@@ -106,6 +106,8 @@ export async function verifyCredentials(db: mongo.Db, cipher: Cipher) {
   }
   let legacy = await db.collection("users").countDocuments({ accessTokens: { $exists: true } });
   for (const name of resources) legacy += await db.collection(name).countDocuments(legacyQuery);
+  const pendingArchiveCleanup = await db.collection("anonymizedrepositories").countDocuments({ archiveCachePending: true });
+  if (pendingArchiveCleanup) throw new Error("Archived repository cache cleanup is pending");
   return { checked, legacy };
 }
 

src/core/model/anonymizedRepositories/anonymizedRepositories.schema.ts

@@ -10,6 +10,9 @@ const AnonymizedRepositorySchema = new Schema({
     default: "preparing",
   },
   statusDate: Date,
+  archivedAt: Date,
+  archiveReason: String,
+  archiveCachePending: Boolean,
   statusMessage: String,
   anonymizeDate: Date,
   lastView: Date,

src/core/model/anonymizedRepositories/anonymizedRepositories.types.ts

@@ -6,6 +6,9 @@ export interface IAnonymizedRepository {
   status?: RepositoryStatus;
   statusMessage?: string;
   statusDate: Date;
+  archivedAt?: Date;
+  archiveReason?: string;
+  archiveCachePending?: boolean;
   anonymizeDate: Date;
   source: {
     type: "GitHubDownload" | "GitHubStream" | "Zip";

src/core/recover-repository-owners.ts

@@ -38,82 +38,159 @@ export async function identifyGitHubToken(token: string): Promise<IdentityResult
 
 export interface RecoveryOptions {
   apply?: boolean;
+  archiveUnrecoverable?: boolean;
+  archiveAllOwnerless?: boolean;
+  concurrency?: number;
   repositoryId?: mongo.ObjectId;
   identify?: (token: string) => Promise<IdentityResult>;
   pause?: () => Promise<void>;
+  deleteCache?: (repoId: string) => Promise<void>;
   report?: (event: { collection: string; id: string; issue?: string;
-    ownerId?: string; githubId?: string; action?: string }) => void;
+    ownerId?: string; githubId?: string; action?: string; reason?: string }) => void;
 }
 
 export async function recoverRepositoryOwners(db: mongo.Db, options: RecoveryOptions = {}) {
+  const concurrency = options.concurrency ?? 5;
+  if (!Number.isInteger(concurrency) || concurrency < 1 || concurrency > 32) {
+    throw new Error("Concurrency must be an integer between 1 and 32");
+  }
   const repositories = db.collection("anonymizedrepositories");
   const users = db.collection("users");
   const identify = options.identify || identifyGitHubToken;
-  const pause = options.pause || (() => new Promise<void>(resolve => setTimeout(resolve, 1000)));
-  // Hashes deduplicate HTTP requests without retaining plaintext as cache keys.
-  const identities = new Map<string, IdentityResult>();
-  const counts = { scanned: 0, candidates: 0, matched: 0, updated: 0, issues: 0, halted: false };
+  const deleteCache = options.deleteCache || (async (repoId: string) => {
+    const storage = (await import("./storage")).default;
+    await storage.rm(repoId);
+  });
+  // Share one pacing gate across workers. At most four new HTTP requests/second.
+  let gate: Promise<void> = Promise.resolve();
+  const pause = options.pause || (() => {
+    const next = gate.then(() => new Promise<void>(resolve => setTimeout(resolve, 250)));
+    gate = next;
+    return next;
+  });
+  const counts = { scanned: 0, candidates: 0, matched: 0, updated: 0,
+    archiveCandidates: 0, archived: 0, cacheDeleted: 0, alreadyArchived: 0, issues: 0, halted: false };
   const report = options.report || (() => {});
-  const query = options.repositoryId ? { _id: options.repositoryId } : {};
-  for await (const row of repositories.find(query, {
-    projection: { owner: 1, "source.accessToken": 1, accessToken: 1 },
-  }).batchSize(100)) {
+  // In-flight lookups are shared too; parallel workers never request the same token twice.
+  const identities = new Map<string, Promise<IdentityResult>>();
+  const lookup = (token: string) => {
+    const hash = createHash("sha256").update(token).digest("hex");
+    let result = identities.get(hash);
+    if (!result) {
+      result = (async () => {
+        await pause();
+        if (counts.halted) return { issue: "scan_halted", halt: true };
+        const identity = await identify(token);
+        if ("issue" in identity && identity.halt) counts.halted = true;
+        return identity;
+      })();
+      identities.set(hash, result);
+      // Evict settled entries only, retaining deduplication of requests in flight.
+      void result.then(() => {
+        if (identities.size > 10000) identities.delete(hash);
+      }, () => {});
+    }
+    return result;
+  };
+  const processRow = async (row: mongo.WithId<mongo.Document>) => {
     counts.scanned++;
-    if (row.owner && await users.findOne({ _id: row.owner }, { projection: { _id: 1 } })) continue;
-    counts.candidates++;
+    if (row.owner && await users.findOne({ _id: row.owner }, { projection: { _id: 1 } })) return;
     const event = { collection: "anonymizedrepositories", id: String(row._id) };
     const fail = (issue: string) => { counts.issues++; report({ ...event, issue }); };
-    const values: unknown[] = [row.source?.accessToken, row.accessToken];
-    if (values.some(value => value != null && typeof value !== "string")) {
-      fail("malformed_token"); continue;
-    }
-    const tokens = [...new Set(values.filter((value): value is string => typeof value === "string" && value.length > 0))];
-    if (!tokens.length) { fail("missing_token"); continue; }
-    const ids = new Set<string>();
-    let failed = false;
-    for (const token of tokens) {
-      const hash = createHash("sha256").update(token).digest("hex");
-      let identity = identities.get(hash);
-      if (!identity) {
-        await pause();
-        identity = await identify(token);
-        if (identities.size >= 10000) identities.clear();
-        identities.set(hash, identity);
+    const unchanged = (value: unknown) => value === undefined ? { $exists: false } : { $eq: value, $exists: true };
+    const original = { _id: row._id, repoId: unchanged(row.repoId), owner: unchanged(row.owner), status: unchanged(row.status),
+      "source.accessToken": unchanged(row.source?.accessToken), accessToken: unchanged(row.accessToken) };
+    const safeRepoId = () => typeof row.repoId === "string" && /^[a-zA-Z0-9_.-]+$/.test(row.repoId) && ![".", ".."].includes(row.repoId);
+    const cleanup = async () => {
+      if (!safeRepoId()) { fail("unsafe_or_missing_repo_id"); return; }
+      try {
+        await deleteCache(row.repoId);
+        const result = await repositories.updateOne({ _id: row._id, status: "archived", archiveCachePending: true }, {
+          $set: { archiveCachePending: false, isReseted: true },
+        });
+        if (!result.modifiedCount) { fail("repository_changed_retry"); return; }
+        counts.cacheDeleted++;
+        report({ ...event, action: "archive_cache_deleted" });
+      } catch {
+        fail("archive_cache_cleanup_failed");
       }
+    };
+    if (row.status === "archived" && row.source?.accessToken === undefined && row.accessToken === undefined) {
+      counts.alreadyArchived++;
+      if (row.archiveCachePending && (options.archiveUnrecoverable || options.archiveAllOwnerless)) {
+        if (options.apply) await cleanup();
+        else report({ ...event, action: "would_delete_archive_cache" });
+      }
+      return;
+    }
+    counts.candidates++;
+    const archive = async (reason: string) => {
+      if (!options.archiveUnrecoverable && !options.archiveAllOwnerless) { fail(reason); return; }
+      if (!safeRepoId()) { fail("unsafe_or_missing_repo_id"); return; }
+      counts.archiveCandidates++;
+      if (options.apply) {
+        const now = new Date();
+        const result = await repositories.updateOne(original, {
+          $set: { status: "archived", statusDate: now, archivedAt: now,
+            archiveReason: reason, archiveCachePending: true, "options.update": false },
+          $unset: { "source.accessToken": "", accessToken: "" },
+        });
+        if (!result.modifiedCount) { fail("repository_changed_retry"); return; }
+        counts.archived++;
+        report({ ...event, reason, action: "archived" });
+        await cleanup();
+      } else report({ ...event, reason, action: "would_archive" });
+    };
+    if (options.archiveAllOwnerless) { await archive("missing_owner"); return; }
+    const values: unknown[] = [row.source?.accessToken, row.accessToken];
+    if (values.some(value => value != null && typeof value !== "string")) { fail("malformed_token"); return; }
+    const tokens = [...new Set(values.filter((value): value is string => typeof value === "string" && value.length > 0))];
+    if (!tokens.length) { await archive("missing_token"); return; }
+    const ids = new Set<string>();
+    let invalidTokens = 0;
+    for (const token of tokens) {
+      const identity = await lookup(token);
       if ("issue" in identity) {
+        if (identity.issue === "invalid_or_revoked_token") { invalidTokens++; continue; }
         fail(identity.issue);
-        counts.halted = !!identity.halt;
-        failed = true;
-        break;
+        return;
       }
       ids.add(identity.githubId);
     }
-    if (counts.halted) break;
-    if (failed) continue;
-    if (ids.size !== 1) { fail("conflicting_token_identities"); continue; }
+    if (invalidTokens === tokens.length) { await archive("invalid_or_revoked_token"); return; }
+    if (invalidTokens) { fail("mixed_token_validity"); return; }
+    if (ids.size !== 1) { fail("conflicting_token_identities"); return; }
     const githubId = [...ids][0];
-    // Match immutable GitHub IDs only. Include historical numeric storage.
     const matches = await users.find({ "externalIDs.github": { $in: [githubId, Number(githubId)] } }, {
       projection: { _id: 1, status: 1 },
     }).limit(2).toArray();
-    if (matches.length !== 1) {
-      fail(matches.length ? "ambiguous_user" : "user_not_found"); continue;
-    }
+    if (matches.length !== 1) { fail(matches.length ? "ambiguous_user" : "user_not_found"); return; }
     const user = matches[0];
-    if (user.status === "removed" || user.status === "banned") { fail("disabled_user"); continue; }
+    if (user.status === "removed" || user.status === "banned") { fail("disabled_user"); return; }
     counts.matched++;
     if (options.apply) {
-      const unchanged = (value: unknown) => value === undefined ? { $exists: false } : { $eq: value, $exists: true };
-      const result = await repositories.updateOne({
-        _id: row._id,
-        owner: unchanged(row.owner),
-        "source.accessToken": unchanged(row.source?.accessToken),
-        accessToken: unchanged(row.accessToken),
-      }, { $set: { owner: user._id } });
-      if (!result.modifiedCount) { fail("repository_changed_retry"); continue; }
+      const result = await repositories.updateOne(original, { $set: { owner: user._id } });
+      if (!result.modifiedCount) { fail("repository_changed_retry"); return; }
       counts.updated++;
     }
     report({ ...event, githubId, ownerId: String(user._id), action: options.apply ? "owner_assigned" : "would_assign_owner" });
+  };
+  const pending = new Set<Promise<void>>();
+  let failed = false;
+  const cursor = repositories.find(options.repositoryId ? { _id: options.repositoryId } : {}, {
+    projection: { owner: 1, repoId: 1, status: 1, archiveCachePending: 1, "source.accessToken": 1, accessToken: 1 },
+  }).batchSize(100);
+  try {
+    for await (const row of cursor) {
+      if (counts.halted || failed) break;
+      const task = processRow(row).catch(() => { failed = true; counts.halted = true; });
+      pending.add(task);
+      void task.then(() => pending.delete(task));
+      if (pending.size >= concurrency) await Promise.race(pending);
+    }
+  } finally {
+    await Promise.all(pending);
   }
+  if (failed) throw new Error("Owner recovery database operation failed");
   return counts;
 }

src/core/types.ts

@@ -9,6 +9,7 @@ export enum RepositoryStatus {
   PREPARING = "preparing",
   DOWNLOAD = "download",
   READY = "ready",
+  ARCHIVED = "archived",
   EXPIRED = "expired",
   EXPIRING = "expiring",
   REMOVED = "removed",