fix: retain recipient suppression after access recovery (#897)

* fix: retain recipient suppression after access recovery

* fix: migrate cold starts and fence resource deactivation

* docs: stop every writer before alert reconciliation

src/core/model/anonymizedGists/anonymizedGists.schema.ts

@@ -1,3 +1,4 @@
+import { ownerAlertLifecycle } from "../owner-alert-lifecycle.schema";
 import { Schema } from "mongoose";
 
 const AnonymizedGistSchema = new Schema({
@@ -74,4 +75,6 @@ const AnonymizedGistSchema = new Schema({
 
 AnonymizedGistSchema.index({ status: 1, "options.expirationDate": 1 });
 
+AnonymizedGistSchema.plugin(ownerAlertLifecycle);
+
 export default AnonymizedGistSchema;

src/core/model/anonymizedPullRequests/anonymizedPullRequests.schema.ts

@@ -1,3 +1,4 @@
+import { ownerAlertLifecycle } from "../owner-alert-lifecycle.schema";
 import { repositoryAccessSchema } from "../repository-access.schema";
 import { Schema } from "mongoose";
 
@@ -71,4 +72,6 @@ const AnonymizedPullRequestSchema = new Schema({
 
 AnonymizedPullRequestSchema.index({ status: 1, "options.expirationDate": 1 });
 
+AnonymizedPullRequestSchema.plugin(ownerAlertLifecycle);
+
 export default AnonymizedPullRequestSchema;

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

@@ -1,3 +1,4 @@
+import { ownerAlertLifecycle } from "../owner-alert-lifecycle.schema";
 import { repositoryAccessSchema } from "../repository-access.schema";
 import { Schema } from "mongoose";
 
@@ -123,4 +124,6 @@ AnonymizedRepositorySchema.index({
   "options.expirationDate": 1,
 });
 
+AnonymizedRepositorySchema.plugin(ownerAlertLifecycle);
+
 export default AnonymizedRepositorySchema;

src/core/model/owner-alert-lifecycle.schema.ts

@@ -0,0 +1,54 @@
+import { randomUUID } from "crypto";
+import { Document, Query, Schema, Types, mongo } from "mongoose";
+
+const inactive = new Set(["archived", "expired", "expiring", "removed", "removing"]);
+type Owner = { _id: Types.ObjectId; accessAlertResetRevision?: number; accessAlertDeactivations?: string[] };
+type Fence = { users: mongo.Collection<Owner>; owners: Types.ObjectId[]; token: string };
+
+/** Lifecycle mutations and email dispatch compete on the owner document. */
+export function ownerAlertLifecycle(schema: Schema): void {
+  const fences = new WeakMap<object, Fence>();
+  const begin = async (operation: object, db: { collection(name: string): unknown }, owners: Types.ObjectId[]) => {
+    if (!owners.length) return;
+    const users = db.collection("users") as mongo.Collection<Owner>;
+    const token = randomUUID();
+    // Do not expire this reservation: an interrupted database mutation may still
+    // commit. Offline reconciliation can clear it after all writers have stopped.
+    await users.updateMany({ _id: { $in: owners } }, {
+      $inc: { accessAlertResetRevision: 1 }, $addToSet: { accessAlertDeactivations: token },
+    });
+    fences.set(operation, { users, owners, token });
+  };
+  const finish = async (operation: object) => {
+    const fence = fences.get(operation);
+    if (!fence) return;
+    await fence.users.updateMany({ _id: { $in: fence.owners } }, { $pull: { accessAlertDeactivations: fence.token } });
+    fences.delete(operation);
+  };
+  schema.pre(["updateOne", "updateMany", "findOneAndUpdate"], async function (this: Query<unknown, unknown>) {
+    const update = this.getUpdate();
+    if (!update) return;
+    const stages = (Array.isArray(update) ? update : [update]) as { $set?: { status?: unknown }; status?: unknown }[];
+    if (!stages.some(stage => { const status = stage.$set?.status ?? stage.status; return typeof status === "string" && inactive.has(status); })) return;
+    const filter = this.getFilter();
+    const lookup = this.model.distinct("owner", filter);
+    const collation = this.getOptions().collation;
+    if (collation) lookup.collation(collation);
+    const owners = (await lookup.exec()).filter(id => id && Types.ObjectId.isValid(id));
+    await begin(this, this.model.db, owners);
+    // Preserve the owner snapshot: reassignment must not move this mutation to
+    // an account whose dispatch fence was not advanced.
+    this.setQuery({ $and: [filter, { $or: [{ owner: { $in: owners } }, { owner: null }] }] });
+  });
+  schema.post(["updateOne", "updateMany", "findOneAndUpdate"], async function () { await finish(this); });
+  schema.pre("save", async function (this: Document & { status?: string; owner?: Types.ObjectId; $where?: Record<string, unknown> }) {
+    if (!inactive.has(this.status || "") || !this.isModified("status")) return;
+    const current = this.isNew ? null : await this.collection.findOne({ _id: this._id }, { projection: { owner: 1 } });
+    const owners = [current?.owner, this.owner].filter((id): id is Types.ObjectId => Boolean(id && Types.ObjectId.isValid(id)));
+    await begin(this, this.db, owners);
+    if (current) this.$where = { $and: [this.$where || {}, { owner: current.owner ?? null }] };
+  });
+  schema.post("save", async function () { await finish(this); });
+  // Failed or ambiguous mutations retain their reservation. Releasing one here
+  // could permit dispatch while the resource write subsequently completes.
+}

src/core/model/users/users.schema.ts

@@ -21,6 +21,7 @@ const UserSchema = new Schema({
   accessAlertResetRevision: { type: Number, select: false },
   accessAlertResetUntil: { type: Date, select: false },
   accessAlertFenceRevision: { type: Number, select: false },
+  accessAlertDeactivations: { type: [String], default: undefined, select: false },
   accessAlertDispatches: { type: Schema.Types.Mixed, select: false },
   emails: [
     {

src/core/owner-notifications.ts

@@ -64,9 +64,10 @@ type AlertRecord = { _id: string; state?: string; lease?: string; leaseUntil?: D
   generation?: string | null; attemptId?: string; claimedAt?: Date; completedAt?: Date;
   recipientKey?: string; resetRevision?: number; httpStatus?: number; startedAt?: Date;
   progress?: Record<string, { claimedAt: Date; id: Types.ObjectId }>; legacyClaims?: LegacyClaim[];
-  resourceRevision?: number; deferredRecoveries?: DeferredRecovery[]; dispatchPrepared?: boolean };
+  resourceRevision?: number; deferredRecoveries?: DeferredRecovery[]; dispatchPrepared?: boolean;
+  deliveryProgress?: { completedAt: Date; id: string }; failureCode?: string };
 const alertCollection = (name = RECIPIENT_ALERTS) => UserModel.db.collection(name) as unknown as mongo.Collection<AlertRecord>;
-const MIGRATION = "recipient-alerts-v1";
+const MIGRATION = "recipient-alerts-v2";
 const PENDING_LEASE_MS = 60000;
 const expiredPending = () => ({ state: "pending", claimedAt: { $lte: new Date(Date.now() - PENDING_LEASE_MS) } });
 const expiredPreparation = () => ({ state: { $in: ["pending", "sending"] }, claimedAt: { $lte: new Date(Date.now() - PENDING_LEASE_MS) } });
@@ -82,6 +83,7 @@ async function prepareOwnerDispatch(record: AlertRecord, revision: number | null
   const path = `accessAlertDispatches.${record.attemptId}`;
   const prepared = await UserModel.db.collection("users").updateOne({ _id: record.owner,
     accessAlertResetRevision: generationFilter(revision),
+    "accessAlertDeactivations.0": { $exists: false },
     $or: [{ accessAlertResetUntil: { $exists: false } }, { accessAlertResetUntil: { $lte: new Date() } }],
   }, [{ $set: { accessAlertDispatches: { $arrayToObject: { $filter: {
     input: { $objectToArray: { $ifNull: ["$accessAlertDispatches", {}] } }, as: "dispatch",
@@ -101,6 +103,7 @@ async function commitOwnerRevision(owner: Types.ObjectId, revision: number | nul
   const path = `accessAlertDispatches.${record?.attemptId}`;
   const committed = await UserModel.db.collection("users").updateOne({ _id: owner,
     accessAlertResetRevision: generationFilter(revision),
+    "accessAlertDeactivations.0": { $exists: false },
     ...(record ? { [`${path}.state`]: "prepared" } : {}),
     $or: [{ accessAlertResetUntil: { $exists: false } }, { accessAlertResetUntil: { $lte: new Date() } }],
   }, record ? { $set: { [`${path}.state`]: "committed" } } : { $inc: { accessAlertFenceRevision: 1 } });
@@ -152,11 +155,14 @@ async function reconcileUnsentAlerts(writersStopped = false): Promise<void> {
   }
 }
 
-/** Call only after all legacy application writers have stopped. */
+/** Call only after all application writers have stopped. */
 export async function prepareLegacyOwnerAlertMigration(): Promise<void> {
   await reconcileUnsentAlerts(true);
+  // Interrupted lifecycle mutations cannot resume once every writer is stopped.
+  await UserModel.db.collection("users").updateMany({ accessAlertDeactivations: { $exists: true } },
+    { $unset: { accessAlertDeactivations: "" }, $inc: { accessAlertResetRevision: 1 } });
   await alertCollection().updateOne({ _id: MIGRATION }, { $set: { state: "new" },
-    $unset: { progress: "", lease: "", leaseUntil: "" } }, { upsert: true });
+    $unset: { progress: "", deliveryProgress: "", lease: "", leaseUntil: "" } }, { upsert: true });
 }
 
 async function reconcileCompletedSend(record: AlertRecord): Promise<void> {
@@ -219,6 +225,43 @@ export async function initializeOwnerAccessAlerts(): Promise<void> {
         .catch(error => { heartbeatError = error; });
     }, 5000);
     try {
+      // Delivery evidence survives v1 guard recovery and resource deletion. Restore
+      // missing guards once, before traffic, so those addresses cannot be emailed again.
+      const events = alertCollection("owner_access_alert_events");
+      const index = "owner_access_alert_delivery_history";
+      await events.createIndex({ completedAt: 1, _id: 1 }, { name: index,
+        partialFilterExpression: { completedAt: { $exists: true } } });
+      const checkpoint = (await alerts.findOne({ _id: MIGRATION, lease }))?.deliveryProgress;
+      const filter = { state: { $in: ["accepted", "failed"] }, completedAt: { $exists: true },
+        ...(checkpoint ? { $or: [{ completedAt: { $gt: checkpoint.completedAt } },
+          { completedAt: checkpoint.completedAt, _id: { $gt: checkpoint.id } }] } : {}) };
+      let lastDelivery: { completedAt: Date; id: string } | undefined;
+      let delivered = 0;
+      const saveDeliveries = async () => {
+        if (!lastDelivery) return;
+        const saved = await alerts.updateOne({ _id: MIGRATION, lease }, { $set: { deliveryProgress: lastDelivery } });
+        if (!saved.matchedCount) throw new Error("owner_alert_migration_lease_lost");
+      };
+      for await (const event of events.find(filter).sort({ completedAt: 1, _id: 1 }).hint(index).batchSize(100)) {
+        if (heartbeatError) throw heartbeatError;
+        if (event.recipientKey && /^[0-9a-f]{64}$/.test(event.recipientKey)) {
+          const previous = await alerts.findOne({ _id: event.recipientKey });
+          if (previous && (["pending", "cancelled"].includes(previous.state || "")
+            || (previous.state === "sending" && previous.dispatchPrepared))) {
+            // A v1 repeat may still be preparing after its earlier guard was
+            // released. Cancel known-unsent work even while its lease is fresh;
+            // committed or ambiguous dispatches remain blocked instead.
+            await releaseUnsentAlert(previous, { state: previous.state });
+          }
+          await alerts.updateOne({ _id: event.recipientKey }, { $setOnInsert: {
+            state: "recipient_notified", attemptId: event.attemptId ?? randomUUID(),
+            claimedAt: event.completedAt, completedAt: event.completedAt, httpStatus: event.httpStatus,
+          } }, { upsert: true });
+        }
+        lastDelivery = { completedAt: event.completedAt!, id: event._id };
+        if (++delivered % 100 === 0) await saveDeliveries();
+      }
+      await saveDeliveries();
       for (const [kind, name] of Object.entries(resourceCollections)) {
         const resources = UserModel.db.collection(name);
         const index = "owner_access_alert_backfill";
@@ -293,7 +336,8 @@ async function completeRecovery(candidate: AlertRecord, recovery: DeferredRecove
   } else if (candidate.state === "pending") {
     await releaseUnsentAlert(candidate, { ...fence, state: "pending" });
   } else {
-    await alertCollection().deleteOne(fence);
+    // Recovery does not authorize another generic email to the same address.
+    await alertCollection().updateOne(fence, { $set: { state: "recovered" } });
   }
 }
 
@@ -337,7 +381,7 @@ async function applyDeferredRecoveries(key: string, attemptId?: string): Promise
   }
 }
 
-/** Only a fresh upstream read after owner action may reopen a recipient alert. */
+/** Confirm recovery after owner action; only known-unsent claims may be released. */
 export async function withOwnerAccessRecovery<T>(ownerId: string, resource: AlertResource, work: () => Promise<T>, expectedSource: Record<string, unknown> = {}, confirmed: (result: T) => boolean = () => true): Promise<T> {
   let snapshot: { accessAlertGeneration?: string } | null = null;
   let candidates: AlertRecord[] = [];
@@ -436,10 +480,11 @@ export async function notifyOwnerAccessProblem(ownerId: string, error: unknown,
   try {
     if (!resource || !Types.ObjectId.isValid(resource.id) || !Types.ObjectId.isValid(ownerId)) return;
     const collection = UserModel.db.collection(resourceCollections[resource.kind]);
-    const identity = { _id: new Types.ObjectId(resource.id), owner: new Types.ObjectId(ownerId) };
+    const identity = { _id: new Types.ObjectId(resource.id), owner: new Types.ObjectId(ownerId),
+      status: { $nin: ["archived", "expired", "expiring", "removed", "removing"] } };
     const ownerState = await UserModel.db.collection("users").findOne({ _id: identity.owner },
-      { projection: { accessAlertResetRevision: 1, accessAlertResetUntil: 1 } });
-    if (!ownerState || ownerState.accessAlertResetUntil?.getTime() > Date.now()) return;
+      { projection: { accessAlertResetRevision: 1, accessAlertResetUntil: 1, accessAlertDeactivations: 1 } });
+    if (!ownerState || ownerState.accessAlertDeactivations?.length || ownerState.accessAlertResetUntil?.getTime() > Date.now()) return;
     const ownerRevision = ownerState.accessAlertResetRevision ?? null;
     const snapshot = await collection.findOne(identity, { projection: { accessAlertGeneration: 1, accessAlertClaimedAt: 1, accessAlertClaimRevision: 1 } });
     if (!snapshot) return;
@@ -486,16 +531,9 @@ export async function notifyOwnerAccessProblem(ownerId: string, error: unknown,
           }
           retired = await releaseUnsentAlert(previous, previous.state === "cancelled" ? retirement : { ...retirement, ...expiredPending() });
         } else {
-          const claims = previous.state === "recovered" ? [] : previous.legacyClaims?.filter(claim => !claim.recovered) ?? [previous as LegacyClaim];
-          for (const claim of claims) {
-            const previousResource = await UserModel.db.collection(resourceCollections[claim.resourceKind]).findOne(
-              { _id: claim.resourceId, owner: claim.owner }, { projection: { status: 1 } });
-            const previousOwner = await UserModel.findById(claim.owner).select("status").lean();
-            if (previousOwner && !["banned", "removed"].includes(previousOwner.status || "")
-              && previousResource && !["removed", "expired"].includes(previousResource.status)) return;
-          }
-          if (previous.legacyClaims) retirement.legacyClaims = previous.legacyClaims;
-          retired = Boolean((await alerts.deleteOne(retirement)).deletedCount);
+          // Delivered, legacy, recovered and ambiguous sends stay blocked permanently,
+          // even if their original resource or account no longer exists.
+          return;
         }
         if (!retired) return;
       }
@@ -516,7 +554,10 @@ export async function notifyOwnerAccessProblem(ownerId: string, error: unknown,
     const base = /^https?:\/\//.test(config.APP_HOSTNAME) ? config.APP_HOSTNAME : `https://${config.APP_HOSTNAME}`;
     const url = new URL("/connections", base).href;
     await alertCollection("owner_access_alert_events").insertOne({ _id: attemptId, attemptId,
-      recipientKey: key, state: "prepared", startedAt: new Date() });
+      recipientKey: key, state: "prepared", startedAt: new Date(), owner: identity.owner,
+      resourceKind: resource.kind, resourceId: identity._id, generation: record.generation,
+      failureCode: accessErrors.has(typeof error === "string" ? error : error instanceof Error ? error.message : "")
+        ? typeof error === "string" ? error : (error as Error).message : "upstream_access_failure" });
     const claim = await collection.updateOne({ ...identity,
       accessAlertGeneration: generationFilter(record.generation),
       accessAlertClaimRevision: revisionFilter(record.resourceRevision ?? 0),
@@ -557,7 +598,8 @@ export async function notifyOwnerAccessProblem(ownerId: string, error: unknown,
     try {
       await retryAlertWrite(async () => {
         try { await alertCollection("owner_access_alert_events").insertOne({ _id: `${attemptId}/completion`, attemptId,
-          recipientKey: key, state, httpStatus, completedAt }); }
+          recipientKey: key, state, httpStatus, completedAt, owner: identity.owner,
+          resourceKind: resource.kind, resourceId: identity._id, generation: record.generation }); }
         catch (error) { if ((error as { code?: number }).code !== 11000) throw error; }
       });
     } catch { logger.warn("owner access email audit unavailable", { attemptId }); }