From 445a479339fba52612ea25e6acf53f6b2aed2b9f Mon Sep 17 00:00:00 2001 From: Thomas Durieux <5577568+tdurieux@users.noreply.github.com> Date: Tue, 13 Oct 2026 14:55:10 +0200 Subject: [PATCH] fix: deduplicate owner access alerts per recipient (#896) * fix: prevent duplicate access alerts to the same recipient * fix: address recipient alert recovery review findings * fix: reconcile unsent alerts and wait for startup migration * fix: preserve legacy recipient alerts through upgrades * fix: reconcile alert results and legacy recovery races * fix: fence dispatch and clean cancelled alert metadata * fix: fence alert preparation and retain in-flight recovery * fix: atomically commit alert dispatch and retry deferred recovery * fix: keep recovery lookups from blocking upstream file reads --- Dockerfile | 1 + README.md | 71 +- scripts/deploy.sh | 38 + src/core/Gist.ts | 6 +- src/core/PullRequest.ts | 7 +- src/core/Repository.ts | 30 +- .../anonymizedGists/anonymizedGists.schema.ts | 2 + .../anonymizedPullRequests.schema.ts | 2 + .../anonymizedRepositories.schema.ts | 2 + src/core/model/users/users.schema.ts | 4 + src/core/owner-notifications.ts | 519 +++++++++++- src/core/source/GitHubStream.ts | 23 +- src/scripts/reconcile-legacy-owner-alerts.ts | 17 + src/server/database.ts | 2 + src/server/routes/connection.ts | 2 +- src/server/routes/github-app.ts | 2 +- test/deploy-owner-alert-migration.test.js | 101 +++ test/owner-notifications.test.js | 227 +++++- test/recipient-alerts-mongo.test.js | 742 ++++++++++++++++++ test/registration-email.test.js | 15 +- 20 files changed, 1724 insertions(+), 89 deletions(-) create mode 100644 src/scripts/reconcile-legacy-owner-alerts.ts create mode 100644 test/deploy-owner-alert-migration.test.js create mode 100644 test/recipient-alerts-mongo.test.js diff --git a/Dockerfile b/Dockerfile index d7f8a39..a34640b 100644 --- a/Dockerfile +++ b/Dockerfile @@ -26,6 +26,7 @@ RUN --mount=type=cache,id=npm-prod,target=/root/.npm,sharing=locked \ # Stage 4: minimal runtime FROM node:22-alpine AS runtime +LABEL science.4open.owner-alert-protocol="recipient-v1" ENV NODE_ENV=production ENV PORT=5000 EXPOSE 5000 diff --git a/README.md b/README.md index c070301..64ad28f 100644 --- a/README.md +++ b/README.md @@ -116,14 +116,69 @@ Alerts cover invalid GitHub credentials and repository access errors encountered while reading or refreshing content. No periodic access scan is performed. Emails contain no repository names, anonymous URLs, or credentials. -An atomic database claim allows one email attempt per repository, pull request, -or gist until its owner explicitly refreshes it, saves its settings or connection, -or reconnects GitHub. Claiming ownership also clears the previous owner's alert. -Reconnecting GitHub or completing a verified App installation update resets alerts -for that owner's resources. Resets invalidate pending claims from older failures. -Viewing content, scheduled updates, and automatic token renewal never reset alerts. -There is no daily reminder. Claims also remain after delivery failures or an -interrupted process, avoiding duplicate messages until the owner acts. +An atomic recipient claim allows one email attempt across repositories, pull +requests, gists and accounts sharing the same address. Sign-in does not reopen +alerts. An explicit reconnect or resource change prepares recovery, but the +recipient stays blocked until a fresh GitHub read confirms access to the resource +that triggered the alert. A fresh read during an in-flight send is retained and +applied when that send completes. A newer reset or source change invalidates +that recovery evidence. Cached reads and successful reads of other resources +cannot reopen it. Removed or expired resources and disabled owners no longer +block future alerts after their send has completed. + +Dispatched recipient guards have no time expiry, so there are no daily reminders. +Failed deliveries remain blocked until confirmed recovery. An interrupted send +whose outcome is unknown stays blocked to avoid duplicate delivery. Unsent claims +can be cancelled by confirmed recovery before dispatch. Bookkeeping failures +release claims when no provider request has started. Pending claims interrupted +before dispatch can be reclaimed after a 60-second lease during startup or a new +notification attempt. Claim identity and state checks prevent a paused sender +from dispatching after another instance reclaims its claim. Cancellation is saved +before cleanup; matching legacy fields are cleared before the guard is released. +Interrupted cleanup resumes on startup or a later notification. Legacy fields +carry an attempt ID so unrelated deliveries and newer generations are preserved. +A resource revision prevents delayed preparation from recreating a reclaimed +claim. All asynchronous preparation stays pending. A sending reservation is +reclaimable until dispatch commits atomically on the owner document, against its +reset revision and lease, immediately before contacting the provider. Cancellation +competes with that same commitment, so a paused sender cannot dispatch afterward. +Owner dispatch records distinguish known-unsent reservations from interrupted +committed sends. Completed records are removed; expired uncommitted records are +pruned on later attempts. Persistent reset revisions also fence attempts created +while a reset is in progress, even when no recipient guard existed at its start. +Recovery recorded during an active reset is retried when reset finishes or its +lease expires, with the original source and revision checks. +Provider idempotency keys +and redacted attempt records support auditing without storing recipient addresses +or repository names in the audit log. Attempt records are saved before dispatch, +with completion recorded separately. Existing resource claims are migrated to +recipient guards before new application instances serve requests using an indexed +scan with resumable checkpoints. Other instances wait while its lease is healthy, +and take over if the lease expires. +Every existing resource claim is retained behind the recipient guard. Recovering +one resource atomically marks its claim recovered; other claims keep the recipient +blocked until they recover or are retired. A newer owner reset makes its matching +claims require another fresh read. Completed provider results are recorded before +guard updates, with bounded retries and reconciliation on startup or a later +notification. Retirement also checks the latest owner reset revision. + +Use `scripts/deploy.sh` for the first upgrade to recipient alerts. The new image +has the `science.4open.owner-alert-protocol=recipient-v1` label. When running +application containers lack that label, the script builds the replacement, stops +and verifies all old API and streamer containers, then prepares a full claim +reconciliation and starts the new streamers and API. Stopped legacy containers +from a rollback also trigger reconciliation, so claims written after an earlier +migration are included. This path also runs when retrying with an already-built +image and `FORCE=0`. Once writers are stopped, preparation releases all known +unsent pending claims and clears their matching legacy fields before backfill. +This upgrade pauses traffic while claims migrate and allows up to 15 +minutes for startup, configurable with `MIGRATION_HEALTH_TIMEOUT`. Subsequent +deployments keep the usual rolling streamer replacement. Manual upgrades must +stop all legacy application writers, run +`docker compose run --rm --no-deps --entrypoint node anonymous_github ./build/scripts/reconcile-legacy-owner-alerts.js` +using the new image, and then start the new application instances. Run this +preparation after legacy writers have operated, even if their containers were +already removed. Ordinary restarts preserve confirmed recoveries. Leaving either Resend setting empty disables delivery and the email prompt. ## Scope of anonymization diff --git a/scripts/deploy.sh b/scripts/deploy.sh index ff1e9f2..8c5cc2a 100755 --- a/scripts/deploy.sh +++ b/scripts/deploy.sh @@ -1,6 +1,7 @@ #!/usr/bin/env bash # Zero-downtime redeploy: # build → blue/green streamer swap → recreate API. +# Upgrading legacy owner alerts pauses application containers before backfill. # Skips everything if the rebuilt image hasn't changed (FORCE=1 to override). set -euo pipefail cd "$(dirname "$0")/.." @@ -35,12 +36,49 @@ list_streamers() { docker compose ps --status running --format '{{.Name}}' streamer 2>/dev/null } +# Capture application writers, including stopped containers from a rollback. +# Legacy instances do not participate in recipient claims, so they must stop +# before a new instance marks the one-time claim migration complete. +application_ids="$(docker compose ps --all -q anonymous_github streamer)" +apps=() +if [[ -n $application_ids ]]; then mapfile -t apps <<<"$application_ids"; fi +legacy_alerts=0 +for app in "${apps[@]}"; do + protocol="$(docker inspect -f '{{index .Config.Labels "science.4open.owner-alert-protocol"}}' "$app")" + [[ $protocol == recipient-v1 ]] || legacy_alerts=1 +done + # 1. Build log "Building image" before="$(image_id "$IMAGE")" docker compose build anonymous_github after="$(image_id "$IMAGE")" +if [[ $legacy_alerts == 1 ]]; then + protocol="$(docker image inspect -f '{{index .Config.Labels "science.4open.owner-alert-protocol"}}' "$IMAGE")" + [[ $protocol == recipient-v1 ]] || { warn "replacement image lacks recipient alert migration support"; exit 1; } + TIMEOUT="${MIGRATION_HEALTH_TIMEOUT:-900}" + log "Pausing legacy application writers before recipient alert migration" + docker compose stop -t 30 anonymous_github streamer + for app in "${apps[@]}"; do + running="$(docker inspect -f '{{.State.Running}}' "$app")" + [[ $running == false ]] || { warn "legacy writer $app is still running; aborting migration"; exit 1; } + done + log "Preparing claim reconciliation, including claims written after rollback" + docker compose run --rm --no-deps --entrypoint node anonymous_github ./build/scripts/reconcile-legacy-owner-alerts.js + log "Starting ${REPLICAS} streamers after legacy writers stopped" + docker compose up -d --no-deps --force-recreate --scale "streamer=${REPLICAS}" streamer + mapfile -t news < <(list_streamers) + [[ ${#news[@]} == "$REPLICAS" ]] || { warn "streamer replica count does not match ${REPLICAS}"; exit 1; } + for app in "${news[@]}"; do wait_healthy "$app"; done + docker compose up -d --no-deps --force-recreate anonymous_github + api="$(docker compose ps -q anonymous_github)" + [[ -n $api ]] || { warn "API container missing after migration"; exit 1; } + wait_healthy "$api" + docker compose ps + exit 0 +fi + # 2. Cold start mapfile -t olds < <(list_streamers) if [[ ${#olds[@]} -eq 0 ]]; then diff --git a/src/core/Gist.ts b/src/core/Gist.ts index 0f9dd34..e638dc7 100644 --- a/src/core/Gist.ts +++ b/src/core/Gist.ts @@ -1,4 +1,4 @@ -import { notifyOwnerAccessProblem } from "./owner-notifications"; +import { notifyOwnerAccessProblem, withOwnerAccessRecovery } from "./owner-notifications"; import { registerGitHubToken } from "./github-token-context"; import { isConnected } from "../server/database"; import { expireEmbeddedContent } from "./content-expiration"; @@ -68,7 +68,9 @@ export default class Gist { } async download() { - try { await this.downloadWithAccess(); } + try { await withOwnerAccessRecovery(this.owner.id, + { kind: "gist", id: String(this.model._id) }, () => this.downloadWithAccess(), + { "source.gistId": this.model.source.gistId }); } catch (error) { void notifyOwnerAccessProblem(this.owner.id, error, { kind: "gist", id: String(this.model._id) }); throw error; diff --git a/src/core/PullRequest.ts b/src/core/PullRequest.ts index 35b4198..3662479 100644 --- a/src/core/PullRequest.ts +++ b/src/core/PullRequest.ts @@ -1,4 +1,4 @@ -import { notifyOwnerAccessProblem } from "./owner-notifications"; +import { notifyOwnerAccessProblem, withOwnerAccessRecovery } from "./owner-notifications"; import { isConnected } from "../server/database"; import { expireEmbeddedContent } from "./content-expiration"; import { boundAppToken } from "./github-app"; @@ -34,7 +34,10 @@ export default class PullRequest { } async download() { - try { await this.downloadContent(); } + try { await withOwnerAccessRecovery(this.owner.id, + { kind: "pull-request", id: String(this.model._id) }, () => this.downloadContent(), + { "source.repositoryFullName": this.model.source.repositoryFullName, "source.pullRequestId": this.model.source.pullRequestId, + "githubAccess.revision": this.model.githubAccess?.revision ?? { $exists: false } }); } catch (error) { void notifyOwnerAccessProblem(this.owner.id, error, { kind: "pull-request", id: String(this.model._id) }); throw error; diff --git a/src/core/Repository.ts b/src/core/Repository.ts index 5e09922..a2efb18 100644 --- a/src/core/Repository.ts +++ b/src/core/Repository.ts @@ -1,4 +1,4 @@ -import { notifyOwnerAccessProblem } from "./owner-notifications"; +import { notifyOwnerAccessProblem, withOwnerAccessRecovery } from "./owner-notifications"; import AnonymizedPathModel from "./model/anonymized-path"; import storage from "./storage"; import { contentGenerationPrefix, contentRetirementMarker } from "./content-generation"; @@ -224,7 +224,11 @@ export default class Repository { const previousPrefix = contentGenerationPrefix(`${previousGeneration || "legacy"}:${this.model.anonymizeDate?.toISOString() || ""}`); const streamed = this.model.source.type !== "Zip"; const source = this.source; - const files = await source.getFiles(opt.progress).catch(error => { + const readFiles = () => source.getFiles(opt.progress); + const files = await (streamed ? withOwnerAccessRecovery(this.owner.id, + { kind: "repository", id: String(this.model._id) }, readFiles, + { "source.repositoryName": this.model.source.repositoryName, "source.commit": this.model.source.commit, + "githubAccess.revision": this.model.githubAccess?.revision ?? { $exists: false } }) : readFiles()).catch(error => { void notifyOwnerAccessProblem(this.owner.id, error, { kind: "repository", id: String(this.model._id) }); throw error; }); @@ -683,12 +687,14 @@ export default class Repository { this.checkRefreshWrite(result); } } - const branches = await ghRepo.branches({ - force: true, - accessToken: token, - }); - const branchName = - this.model.source.branch || ghRepo.model.defaultBranch; + const branchName = this.model.source.branch || ghRepo.model.defaultBranch; + const branches = await withOwnerAccessRecovery(this.owner.id, + { kind: "repository", id: String(this.model._id) }, + () => ghRepo.branches({ force: true, accessToken: token }), + { "source.repositoryName": this.model.source.repositoryName, + "githubAccess.revision": this.model.githubAccess?.revision ?? { $exists: false } }, + result => !opt?.force && this.status === RepositoryStatus.READY + && result.some(branch => branch.name === branchName && branch.commit === this.model.source.commit)); const newCommit = branches.filter((f) => f.name == branchName)[0] ?.commit; @@ -716,9 +722,11 @@ export default class Repository { return; } this._model.source.commit = newCommit; - const commitInfo = await ghRepo.getCommitInfo(newCommit, { - accessToken: token, - }); + const commitInfo = await withOwnerAccessRecovery(this.owner.id, + { kind: "repository", id: String(this.model._id) }, + () => ghRepo.getCommitInfo(newCommit, { accessToken: token }), + { "source.repositoryName": this.model.source.repositoryName, + "githubAccess.revision": this.model.githubAccess?.revision ?? { $exists: false } }); if ( commitInfo.commit?.author?.date || commitInfo.commit?.committer?.date diff --git a/src/core/model/anonymizedGists/anonymizedGists.schema.ts b/src/core/model/anonymizedGists/anonymizedGists.schema.ts index 65f06f1..920f57c 100644 --- a/src/core/model/anonymizedGists/anonymizedGists.schema.ts +++ b/src/core/model/anonymizedGists/anonymizedGists.schema.ts @@ -3,6 +3,8 @@ import { Schema } from "mongoose"; const AnonymizedGistSchema = new Schema({ accessAlertGeneration: { type: String, select: false }, accessAlertClaimedAt: { type: Date, select: false }, + accessAlertAttemptId: { type: String, select: false }, + accessAlertClaimRevision: { type: Number, select: false }, gistId: { type: String, index: { unique: true }, diff --git a/src/core/model/anonymizedPullRequests/anonymizedPullRequests.schema.ts b/src/core/model/anonymizedPullRequests/anonymizedPullRequests.schema.ts index 8ae76ed..ff45881 100644 --- a/src/core/model/anonymizedPullRequests/anonymizedPullRequests.schema.ts +++ b/src/core/model/anonymizedPullRequests/anonymizedPullRequests.schema.ts @@ -4,6 +4,8 @@ import { Schema } from "mongoose"; const AnonymizedPullRequestSchema = new Schema({ accessAlertGeneration: { type: String, select: false }, accessAlertClaimedAt: { type: Date, select: false }, + accessAlertAttemptId: { type: String, select: false }, + accessAlertClaimRevision: { type: Number, select: false }, pullRequestId: { type: String, index: { unique: true }, diff --git a/src/core/model/anonymizedRepositories/anonymizedRepositories.schema.ts b/src/core/model/anonymizedRepositories/anonymizedRepositories.schema.ts index cd5acf9..ff8e19b 100644 --- a/src/core/model/anonymizedRepositories/anonymizedRepositories.schema.ts +++ b/src/core/model/anonymizedRepositories/anonymizedRepositories.schema.ts @@ -4,6 +4,8 @@ import { Schema } from "mongoose"; const AnonymizedRepositorySchema = new Schema({ accessAlertGeneration: { type: String, select: false }, accessAlertClaimedAt: { type: Date, select: false }, + accessAlertAttemptId: { type: String, select: false }, + accessAlertClaimRevision: { type: Number, select: false }, reviewConsentRevision: { type: Number, select: false }, repoId: { type: String, diff --git a/src/core/model/users/users.schema.ts b/src/core/model/users/users.schema.ts index 63c77c3..27589a1 100644 --- a/src/core/model/users/users.schema.ts +++ b/src/core/model/users/users.schema.ts @@ -18,6 +18,10 @@ const UserSchema = new Schema({ notificationEmail: { type: String }, emailPromptNever: { type: Boolean, default: false }, accessAlertAfter: { type: Date, select: false }, + accessAlertResetRevision: { type: Number, select: false }, + accessAlertResetUntil: { type: Date, select: false }, + accessAlertFenceRevision: { type: Number, select: false }, + accessAlertDispatches: { type: Schema.Types.Mixed, select: false }, emails: [ { email: { type: String }, diff --git a/src/core/owner-notifications.ts b/src/core/owner-notifications.ts index 1122277..ed11b09 100644 --- a/src/core/owner-notifications.ts +++ b/src/core/owner-notifications.ts @@ -1,5 +1,5 @@ -import { randomUUID } from "crypto"; -import { Types } from "mongoose"; +import { createHash, randomUUID } from "crypto"; +import { mongo, Types } from "mongoose"; import { isGitHubRateLimitError } from "./github-rate-limit"; import config from "../config"; import UserModel from "./model/users/users.model"; @@ -17,13 +17,378 @@ export interface AlertResource { kind: keyof typeof resourceCollections; id: str export async function resetOwnerAccessAlerts(ownerId: string, resource?: AlertResource): Promise { if (UserModel.db.readyState !== 1 || !Types.ObjectId.isValid(ownerId)) return; try { - const filter = { owner: new Types.ObjectId(ownerId), - ...(resource ? { _id: new Types.ObjectId(resource.id) } : {}) }; - for (const collection of resource ? [resourceCollections[resource.kind]] : Object.values(resourceCollections)) { - await UserModel.db.collection(collection).updateMany(filter, { $unset: { accessAlertClaimedAt: "" }, $set: { accessAlertGeneration: randomUUID() } }); + const users = UserModel.db.collection("users"); + const owner = new Types.ObjectId(ownerId); + const reset = await users.findOneAndUpdate({ _id: owner }, { $inc: { accessAlertResetRevision: 1 }, + $set: { accessAlertResetUntil: new Date(Date.now() + 60000) } }, { returnDocument: "after" }); + if (!reset.value) return; + const revision = reset.value.accessAlertResetRevision; + const heartbeat = setInterval(() => { + void users.updateOne({ _id: owner, accessAlertResetRevision: revision, accessAlertResetUntil: { $exists: true } }, + { $set: { accessAlertResetUntil: new Date(Date.now() + 60000) } }).catch(() => {}); + }, 10000); + try { + const filter = { owner: new Types.ObjectId(ownerId), + ...(resource ? { _id: new Types.ObjectId(resource.id) } : {}) }; + const claim = { owner: new Types.ObjectId(ownerId), + ...(resource ? { resourceKind: resource.kind, resourceId: new Types.ObjectId(resource.id) } : {}) }; + await alertCollection().updateMany({ ...claim, legacyClaims: { $exists: false } }, { $inc: { resetRevision: 1 } }); + await alertCollection().updateMany({ legacyClaims: { $elemMatch: claim } }, [{ $set: { + resetRevision: { $add: [{ $ifNull: ["$resetRevision", 0] }, 1] }, + legacyClaims: { $map: { input: "$legacyClaims", as: "claim", in: { $cond: [ + { $and: Object.entries(claim).map(([key, value]) => ({ $eq: [`$$claim.${key}`, value] })) }, + { $mergeObjects: ["$$claim", { recovered: false }] }, "$$claim", + ] } } }, + state: { $cond: [{ $eq: ["$state", "recovered"] }, "legacy", "$state"] }, + } }]); + for (const collection of resource ? [resourceCollections[resource.kind]] : Object.values(resourceCollections)) { + await UserModel.db.collection(collection).updateMany(filter, { $unset: { accessAlertClaimedAt: "", accessAlertAttemptId: "" }, $set: { accessAlertGeneration: randomUUID() } }); + } + } finally { + clearInterval(heartbeat); + const finished = await users.updateOne({ _id: owner, accessAlertResetRevision: revision }, { $unset: { accessAlertResetUntil: "" } }); + if (finished.matchedCount) { + for await (const candidate of alertCollection().find({ deferredRecoveries: { $elemMatch: { owner } } })) { + await applyDeferredRecoveries(candidate._id, candidate.attemptId); + } + } } } catch { logger.warn("owner access alert reset unavailable"); } } +export const RECIPIENT_ALERTS = "owner_access_alert_recipients"; +type LegacyClaim = { owner: Types.ObjectId; resourceKind: AlertResource["kind"]; resourceId: Types.ObjectId; generation: string | null; recovered?: boolean }; +type DeferredRecovery = { owner: Types.ObjectId; resourceKind: AlertResource["kind"]; resourceId: Types.ObjectId; + generation: string | null; resetRevision: number | null; ownerRevision: number | null; bindingFields: string[]; bindingHash: string }; +type AlertRecord = { _id: string; state?: string; lease?: string; leaseUntil?: Date; + owner?: Types.ObjectId; resourceKind?: string; resourceId?: Types.ObjectId; + generation?: string | null; attemptId?: string; claimedAt?: Date; completedAt?: Date; + recipientKey?: string; resetRevision?: number; httpStatus?: number; startedAt?: Date; + progress?: Record; legacyClaims?: LegacyClaim[]; + resourceRevision?: number; deferredRecoveries?: DeferredRecovery[]; dispatchPrepared?: boolean }; +const alertCollection = (name = RECIPIENT_ALERTS) => UserModel.db.collection(name) as unknown as mongo.Collection; +const MIGRATION = "recipient-alerts-v1"; +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) } }); +const revisionFilter = (revision: number) => revision === 0 ? { $in: [0, null] } : revision; +const generationFilter = (generation: unknown) => generation ?? { $exists: false }; +export const recipientAlertKey = (email: string) => createHash("sha256").update(email.trim().toLowerCase()).digest("hex"); +const bindingHash = (row: Record, fields: string[]) => createHash("sha256").update(JSON.stringify(fields.map(field => { + const value = field.split(".").reduce((value, key) => value && typeof value === "object" ? (value as Record)[key] : undefined, row); + return [field, value !== undefined, value]; +}))).digest("hex"); + +async function prepareOwnerDispatch(record: AlertRecord, revision: number | null): Promise { + const path = `accessAlertDispatches.${record.attemptId}`; + const prepared = await UserModel.db.collection("users").updateOne({ _id: record.owner, + accessAlertResetRevision: generationFilter(revision), + $or: [{ accessAlertResetUntil: { $exists: false } }, { accessAlertResetUntil: { $lte: new Date() } }], + }, [{ $set: { accessAlertDispatches: { $arrayToObject: { $filter: { + input: { $objectToArray: { $ifNull: ["$accessAlertDispatches", {}] } }, as: "dispatch", + cond: { $or: [{ $eq: ["$$dispatch.v.state", "committed"] }, + { $gt: ["$$dispatch.v.claimedAt", new Date(Date.now() - PENDING_LEASE_MS)] }] }, + } } } } }, { $set: { [path]: { state: "prepared", claimedAt: record.claimedAt } } }]); + return Boolean(prepared.matchedCount); +} + +async function clearOwnerDispatch(record: AlertRecord, state: string): Promise { + if (!record.dispatchPrepared || !record.owner || !record.attemptId) return; + const path = `accessAlertDispatches.${record.attemptId}`; + await UserModel.db.collection("users").updateOne({ _id: record.owner, [`${path}.state`]: state }, { $unset: { [path]: "" } }); +} + +async function commitOwnerRevision(owner: Types.ObjectId, revision: number | null | undefined, record?: AlertRecord): Promise { + const path = `accessAlertDispatches.${record?.attemptId}`; + const committed = await UserModel.db.collection("users").updateOne({ _id: owner, + accessAlertResetRevision: generationFilter(revision), + ...(record ? { [`${path}.state`]: "prepared" } : {}), + $or: [{ accessAlertResetUntil: { $exists: false } }, { accessAlertResetUntil: { $lte: new Date() } }], + }, record ? { $set: { [`${path}.state`]: "committed" } } : { $inc: { accessAlertFenceRevision: 1 } }); + return Boolean(committed.matchedCount); +} + +async function releaseUnsentAlert(record: AlertRecord, expected: mongo.Filter): Promise { + if (!record.attemptId) return false; + const alerts = alertCollection(); + if (record.dispatchPrepared && record.owner) { + const users = UserModel.db.collection("users"); + const path = `accessAlertDispatches.${record.attemptId}`; + // This CAS competes with the final dispatch commitment on the owner document. + const cancelled = await users.updateOne({ _id: record.owner, [`${path}.state`]: "prepared" }, + { $set: { [`${path}.state`]: "cancelled" } }); + if (!cancelled.matchedCount) { + const owner = await users.findOne({ _id: record.owner }, { projection: { [path]: 1 } }); + if (!owner || owner.accessAlertDispatches?.[record.attemptId]?.state === "committed") return false; + } + } + // A durable cancellation prevents a paused sender from starting dispatch. + const cancelled = await alerts.updateOne({ _id: record._id, attemptId: record.attemptId, ...expected }, { $set: { state: "cancelled" } }); + if (!cancelled.matchedCount) return false; + if (record.owner && record.resourceKind && record.resourceKind in resourceCollections && record.resourceId) { + const resource = UserModel.db.collection(resourceCollections[record.resourceKind as AlertResource["kind"]]); + const revision = record.resourceRevision ?? 0; + const matching = { + _id: record.resourceId, owner: record.owner, accessAlertGeneration: generationFilter(record.generation), + $or: [{ accessAlertClaimRevision: revisionFilter(revision) }, + { accessAlertClaimRevision: revision + 1, accessAlertAttemptId: record.attemptId }], + }; + await retryAlertWrite(() => resource.updateOne(matching, [{ $set: { + accessAlertClaimRevision: { $add: [{ $ifNull: ["$accessAlertClaimRevision", 0] }, 1] }, + accessAlertClaimedAt: { $cond: [{ $eq: ["$accessAlertAttemptId", record.attemptId] }, "$$REMOVE", "$accessAlertClaimedAt"] }, + accessAlertAttemptId: { $cond: [{ $eq: ["$accessAlertAttemptId", record.attemptId] }, "$$REMOVE", "$accessAlertAttemptId"] }, + } }])); + } + const removed = await alerts.deleteOne({ _id: record._id, attemptId: record.attemptId, state: "cancelled" }); + if (removed.deletedCount) await clearOwnerDispatch(record, "cancelled"); + return Boolean(removed.deletedCount); +} + +async function reconcileUnsentAlerts(writersStopped = false): Promise { + const pending = writersStopped ? { state: "pending" } : expiredPending(); + const preparing = { ...(writersStopped ? { state: "sending" } : { ...expiredPreparation(), state: "sending" }), dispatchPrepared: true }; + for await (const record of alertCollection().find({ $or: [pending, preparing, { state: "cancelled" }] })) { + await releaseUnsentAlert(record, record.state === "cancelled" ? { state: "cancelled" } + : writersStopped ? { state: record.state } : expiredPreparation()); + } +} + +/** Call only after all legacy application writers have stopped. */ +export async function prepareLegacyOwnerAlertMigration(): Promise { + await reconcileUnsentAlerts(true); + await alertCollection().updateOne({ _id: MIGRATION }, { $set: { state: "new" }, + $unset: { progress: "", lease: "", leaseUntil: "" } }, { upsert: true }); +} + +async function reconcileCompletedSend(record: AlertRecord): Promise { + if (record.state !== "sending" || !record.attemptId) return; + const completion = await alertCollection("owner_access_alert_events").findOne({ _id: `${record.attemptId}/completion` }); + if (!completion || !["accepted", "failed"].includes(completion.state || "")) return; + await alertCollection().updateOne({ _id: record._id, attemptId: record.attemptId, state: "sending" }, + { $set: { state: completion.state, completedAt: completion.completedAt, httpStatus: completion.httpStatus } }); + await clearOwnerDispatch(record, "committed"); + await applyDeferredRecoveries(record._id, record.attemptId); +} + +async function reconcileCompletedSends(): Promise { + for await (const record of alertCollection().find({ state: "sending" })) await reconcileCompletedSend(record); + for await (const record of alertCollection().find({ state: { $in: ["accepted", "failed", "legacy"] }, + $or: [{ deferredRecoveries: { $exists: true } }, { dispatchPrepared: true }] })) { + await clearOwnerDispatch(record, "committed"); + await applyDeferredRecoveries(record._id, record.attemptId); + } +} + +async function retryAlertWrite(work: () => Promise): Promise { + for (let attempt = 0; ; attempt++) { + try { await work(); return; } + catch (error) { + if (attempt === 2) throw error; + await new Promise(resolve => setTimeout(resolve, 50 * (attempt + 1))); + } + } +} + +/** Seed the recipient guard once before any new application instance serves traffic. */ +export async function initializeOwnerAccessAlerts(): Promise { + const alerts = alertCollection(); + await alerts.createIndex({ owner: 1, resourceKind: 1, resourceId: 1 }); + await alerts.createIndex({ "legacyClaims.owner": 1, "legacyClaims.resourceKind": 1, "legacyClaims.resourceId": 1 }); + await alerts.createIndex({ claimedAt: 1 }, { name: "owner_access_alert_pending", partialFilterExpression: { state: "pending" } }); + await alerts.createIndex({ state: 1 }); + await alerts.createIndex({ "deferredRecoveries.owner": 1 }); + // Clear known-unsent legacy fields before they can be mistaken for deliveries. + await reconcileUnsentAlerts(); + try { await alerts.updateOne({ _id: MIGRATION }, { $setOnInsert: { state: "new" } }, { upsert: true }); } + catch (error) { if ((error as { code?: number }).code !== 11000) throw error; } + const lease = randomUUID(); + for (;;) { + if ((await alerts.findOne({ _id: MIGRATION }))?.state === "ready") { + await reconcileCompletedSends(); + await reconcileUnsentAlerts(); + return; + } + const now = new Date(); + const claim = await alerts.updateOne({ _id: MIGRATION, state: { $ne: "ready" }, + $or: [{ leaseUntil: { $lt: now } }, { leaseUntil: { $exists: false } }], + }, { $set: { lease, leaseUntil: new Date(Date.now() + 60000) } }); + if (!claim.modifiedCount) { await new Promise(resolve => setTimeout(resolve, 250)); continue; } + let heartbeatError: unknown; + const heartbeat = setInterval(() => { + void alerts.updateOne({ _id: MIGRATION, lease }, { $set: { leaseUntil: new Date(Date.now() + 60000) } }) + .then(result => { if (!result.matchedCount) heartbeatError = new Error("owner_alert_migration_lease_lost"); }) + .catch(error => { heartbeatError = error; }); + }, 5000); + try { + for (const [kind, name] of Object.entries(resourceCollections)) { + const resources = UserModel.db.collection(name); + const index = "owner_access_alert_backfill"; + await resources.createIndex({ accessAlertClaimedAt: 1, _id: 1 }, + { name: index, partialFilterExpression: { accessAlertClaimedAt: { $exists: true } } }); + const checkpoint = (await alerts.findOne({ _id: MIGRATION, lease }))?.progress?.[kind]; + const filter = { accessAlertClaimedAt: { $exists: true }, ...(checkpoint ? { $or: [ + { accessAlertClaimedAt: { $gt: checkpoint.claimedAt } }, + { accessAlertClaimedAt: checkpoint.claimedAt, _id: { $gt: checkpoint.id } }, + ] } : {}) }; + const cursor = resources.find(filter, + { projection: { owner: 1, accessAlertGeneration: 1, accessAlertClaimedAt: 1 } }) + .sort({ accessAlertClaimedAt: 1, _id: 1 }).hint(index).batchSize(100); + let count = 0; + let last: { claimedAt: Date; id: Types.ObjectId } | undefined; + const saveProgress = async () => { + if (!last) return; + const saved = await alerts.updateOne({ _id: MIGRATION, lease }, { $set: { [`progress.${kind}`]: last } }); + if (!saved.matchedCount) throw new Error("owner_alert_migration_lease_lost"); + }; + for await (const row of cursor) { + if (heartbeatError) throw heartbeatError; + last = { claimedAt: row.accessAlertClaimedAt, id: row._id }; + const owner = await UserModel.findById(row.owner).select("emails notificationEmail status").lean(); + const email = notificationEmail(owner?.emails, owner?.notificationEmail); + const legacyClaim: LegacyClaim = { owner: row.owner, resourceKind: kind as AlertResource["kind"], + resourceId: row._id, generation: row.accessAlertGeneration ?? null, recovered: false }; + if (email && owner && !["removed", "banned"].includes(owner.status || "")) { + const key = recipientAlertKey(email); + const previous = await alerts.findOne({ _id: key }); + const claims = [legacyClaim]; + if (previous && !previous.legacyClaims && previous.owner && previous.resourceKind && previous.resourceId) { + claims.push({ owner: previous.owner, resourceKind: previous.resourceKind as AlertResource["kind"], + resourceId: previous.resourceId, generation: previous.generation ?? null, recovered: false }); + } + await alerts.updateOne({ _id: key, legacyClaims: { $elemMatch: { owner: legacyClaim.owner, + resourceKind: legacyClaim.resourceKind, resourceId: legacyClaim.resourceId, generation: legacyClaim.generation } } }, + { $set: { "legacyClaims.$.recovered": false } }); + await alerts.updateOne({ _id: key }, { $setOnInsert: { owner: legacyClaim.owner, resourceKind: kind, + resourceId: row._id, generation: legacyClaim.generation, attemptId: randomUUID(), + claimedAt: row.accessAlertClaimedAt, state: "legacy", + }, $addToSet: { legacyClaims: { $each: claims } } }, { upsert: true }); + await alerts.updateOne({ _id: key, state: { $in: ["recovered", "pending"] } }, { $set: { state: "legacy" } }); + } + if (++count % 100 === 0) await saveProgress(); + } + await saveProgress(); + } + if (heartbeatError) throw heartbeatError; + const completed = await alerts.updateOne({ _id: MIGRATION, lease }, { $set: { state: "ready" }, $unset: { lease: "", leaseUntil: "" } }); + if (!completed.matchedCount) throw new Error("owner_alert_migration_lease_lost"); + await reconcileCompletedSends(); + await reconcileUnsentAlerts(); + return; + } finally { clearInterval(heartbeat); } + } +} + +async function completeRecovery(candidate: AlertRecord, recovery: DeferredRecovery): Promise { + if (candidate.state !== "pending") await clearOwnerDispatch(candidate, "committed"); + const fence = { _id: candidate._id, attemptId: candidate.attemptId, resetRevision: generationFilter(recovery.resetRevision), + state: { $in: ["pending", "accepted", "failed", "legacy"] } }; + if (candidate.legacyClaims) { + await alertCollection().updateOne(fence, [{ $set: { legacyClaims: { $map: { + input: "$legacyClaims", as: "claim", in: { $cond: [{ $and: [ + { $eq: ["$$claim.owner", recovery.owner] }, { $eq: ["$$claim.resourceKind", recovery.resourceKind] }, + { $eq: ["$$claim.resourceId", recovery.resourceId] }, { $ne: ["$$claim.generation", recovery.generation] }, + ] }, { $mergeObjects: ["$$claim", { recovered: true }] }, "$$claim"] }, + } } } }, { $set: { state: { $cond: [{ $allElementsTrue: [{ $map: { + input: "$legacyClaims", as: "claim", in: { $eq: ["$$claim.recovered", true] }, + } }] }, "recovered", "$state"] } } }]); + } else if (candidate.state === "pending") { + await releaseUnsentAlert(candidate, { ...fence, state: "pending" }); + } else { + await alertCollection().deleteOne(fence); + } +} + +const delayedRecoveries = new Map(); +function deferRecoveryUntil(key: string, attemptId: string | undefined, until: Date): void { + const timerKey = `${key}/${attemptId}`; + if (delayedRecoveries.has(timerKey)) return; + const timer = setTimeout(() => { + delayedRecoveries.delete(timerKey); + void retryAlertWrite(() => applyDeferredRecoveries(key, attemptId)) + .catch(() => logger.warn("owner access email deferred recovery unavailable")); + }, Math.max(1, until.getTime() - Date.now() + 1)); + timer.unref(); + delayedRecoveries.set(timerKey, timer); +} + +async function applyDeferredRecoveries(key: string, attemptId?: string): Promise { + const timerKey = `${key}/${attemptId}`; + clearTimeout(delayedRecoveries.get(timerKey)); + delayedRecoveries.delete(timerKey); + const candidate = await alertCollection().findOne({ _id: key, attemptId }); + if (!candidate || !["pending", "accepted", "failed", "legacy"].includes(candidate.state || "")) return; + for (const recovery of candidate.deferredRecoveries ?? []) { + if (recovery.resetRevision !== (candidate.resetRevision ?? null)) continue; + const row = await UserModel.db.collection(resourceCollections[recovery.resourceKind]).findOne({ + _id: recovery.resourceId, owner: recovery.owner, accessAlertGeneration: generationFilter(recovery.generation), + }); + if (!row || bindingHash(row, recovery.bindingFields) !== recovery.bindingHash) continue; + if (!await commitOwnerRevision(recovery.owner, recovery.ownerRevision)) { + const owner = await UserModel.db.collection("users").findOne({ _id: recovery.owner, + accessAlertResetRevision: generationFilter(recovery.ownerRevision) }, { projection: { accessAlertResetUntil: 1 } }); + if (!owner) continue; + if (owner.accessAlertResetUntil?.getTime() > Date.now()) { + deferRecoveryUntil(key, attemptId, owner.accessAlertResetUntil); + continue; + } + // The lease may have cleared between the failed CAS and this read. + if (!await commitOwnerRevision(recovery.owner, recovery.ownerRevision)) continue; + } + await completeRecovery(candidate, recovery); + } +} + +/** Only a fresh upstream read after owner action may reopen a recipient alert. */ +export async function withOwnerAccessRecovery(ownerId: string, resource: AlertResource, work: () => Promise, expectedSource: Record = {}, confirmed: (result: T) => boolean = () => true): Promise { + let snapshot: { accessAlertGeneration?: string } | null = null; + let candidates: AlertRecord[] = []; + let ownerRevision: number | null = null; + const valid = UserModel.db.readyState === 1 && Types.ObjectId.isValid(ownerId) && Types.ObjectId.isValid(resource.id); + const identity = valid ? { _id: new Types.ObjectId(resource.id), owner: new Types.ObjectId(ownerId), ...expectedSource } : null; + if (identity) { + try { + const row = await UserModel.db.collection(resourceCollections[resource.kind]).findOne(identity, { projection: { accessAlertGeneration: 1 } }); + snapshot = row ? { accessAlertGeneration: row.accessAlertGeneration } : null; + if (snapshot) { + const claim = { owner: identity.owner, resourceKind: resource.kind, resourceId: identity._id, + generation: { $ne: snapshot.accessAlertGeneration ?? null } }; + candidates = await alertCollection().find({ state: { $in: ["pending", "accepted", "failed", "legacy", "sending"] }, + $or: [{ ...claim, legacyClaims: { $exists: false } }, { legacyClaims: { $elemMatch: { ...claim, recovered: { $ne: true } } } }], + }).toArray(); + if (candidates.length) ownerRevision = (await UserModel.db.collection("users").findOne({ _id: identity.owner }, + { projection: { accessAlertResetRevision: 1 } }))?.accessAlertResetRevision ?? null; + } + } + catch { /* Email bookkeeping must not change upstream behavior. */ } + } + const result = await work(); + if (!identity || !snapshot || !confirmed(result)) return result; + try { + const current = await UserModel.db.collection(resourceCollections[resource.kind]).findOne({ ...identity, + accessAlertGeneration: generationFilter(snapshot.accessAlertGeneration), + }); + if (!current) return result; + let recorded = 0; + for (const candidate of candidates) { + const fence = { _id: candidate._id, attemptId: candidate.attemptId, resetRevision: generationFilter(candidate.resetRevision), + state: { $in: ["pending", "accepted", "failed", "legacy", "sending"] } }; + const fields = Object.keys(expectedSource); + const recovery: DeferredRecovery = { owner: identity.owner, resourceKind: resource.kind, resourceId: identity._id, + generation: snapshot.accessAlertGeneration ?? null, resetRevision: candidate.resetRevision ?? null, ownerRevision, + bindingFields: fields, bindingHash: bindingHash(current, fields) }; + const saved = await alertCollection().updateOne(fence, [{ $set: { deferredRecoveries: { $concatArrays: [ + { $filter: { input: { $ifNull: ["$deferredRecoveries", []] }, as: "recovery", cond: { $not: [{ $and: [ + { $eq: ["$$recovery.owner", identity.owner] }, { $eq: ["$$recovery.resourceKind", resource.kind] }, + { $eq: ["$$recovery.resourceId", identity._id] }, + ] }] } } }, [recovery], + ] } } }]); + if (saved.matchedCount) recorded++; + await applyDeferredRecoveries(candidate._id, candidate.attemptId); + } + if (recorded) logger.info("owner access email recovery recorded", { resources: recorded }); + } catch { logger.warn("owner access email recovery unavailable"); } + return result; +} + const accessErrors = new Set([ "token_expired", "github_oauth_required", "github_app_reconnect_required", "github_app_access_required", "repo_not_found", "repository_not_found", @@ -65,31 +430,116 @@ export function isAccessFailure(error: unknown): boolean { /** Best effort; mail and database failures must not change repository behavior. */ export async function notifyOwnerAccessProblem(ownerId: string, error: unknown, resource: AlertResource): Promise { if (!config.RESEND_API_KEY || !config.EMAIL_FROM || !isAccessFailure(error)) return; + let unsent: AlertRecord | undefined; + let dispatchRecord: AlertRecord | undefined; + let providerStarted = false; 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 snapshot = await collection.findOne(identity, { projection: { accessAlertGeneration: 1 } }); + const ownerState = await UserModel.db.collection("users").findOne({ _id: identity.owner }, + { projection: { accessAlertResetRevision: 1, accessAlertResetUntil: 1 } }); + if (!ownerState || 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; const owner = await UserModel.findOne({ _id: ownerId, status: { $nin: ["removed", "banned"] } }).lean(); if (!owner) return; const to = notificationEmail(owner.emails, owner.notificationEmail); if (!to) return; - // Claim the resource atomically before delivery. This has no expiry: neither - // repeated readers nor workers can send again until the owner acts. - const claim = await collection.updateOne({ - ...identity, - // A reset during the owner lookup invalidates this older failure. - accessAlertGeneration: snapshot.accessAlertGeneration ?? { $exists: false }, - accessAlertClaimedAt: { $exists: false }, - }, { $set: { accessAlertClaimedAt: new Date() } }); - if (!claim.modifiedCount) return; + const alerts = alertCollection(); + const key = recipientAlertKey(to); + const attemptId = randomUUID(); + const record: AlertRecord = { _id: key, owner: identity.owner, resourceKind: resource.kind, + resourceId: identity._id, generation: snapshot.accessAlertGeneration ?? null, + resourceRevision: snapshot.accessAlertClaimRevision ?? 0, attemptId, claimedAt: new Date(), state: "pending", dispatchPrepared: true }; + dispatchRecord = record; + if (!await prepareOwnerDispatch(record, ownerRevision)) return; + unsent = record; + try { await alerts.insertOne(record); } + catch (error) { + if ((error as { code?: number }).code !== 11000) throw error; + unsent = undefined; + let previous = await alerts.findOne({ _id: key }); + if (previous?.state === "sending") { + await reconcileCompletedSend(previous); + previous = await alerts.findOne({ _id: key }); + } + if (previous && ["accepted", "failed"].includes(previous.state || "")) { + await clearOwnerDispatch(previous, "committed"); + await applyDeferredRecoveries(previous._id, previous.attemptId); + previous = await alerts.findOne({ _id: key }); + } + if (previous?.state === "sending" && previous.dispatchPrepared && previous.claimedAt + && previous.claimedAt.getTime() <= Date.now() - PENDING_LEASE_MS) { + if (await releaseUnsentAlert(previous, expiredPreparation())) previous = await alerts.findOne({ _id: key }); + } + if (previous && (previous.state === "sending" || !previous.owner || !previous.resourceId + || !previous.resourceKind || !(previous.resourceKind in resourceCollections))) return; + if (previous) { + const retirement: mongo.Filter = { _id: key, attemptId: previous.attemptId, state: previous.state, + resetRevision: generationFilter(previous.resetRevision) }; + let retired: boolean; + if (previous.state === "pending" || previous.state === "cancelled") { + if (previous.state === "pending") { + if (!previous.claimedAt || previous.claimedAt.getTime() > Date.now() - PENDING_LEASE_MS) return; + } + 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); + } + if (!retired) return; + } + unsent = record; + try { await alerts.insertOne(record); } + catch (replacementError) { if ((replacementError as { code?: number }).code === 11000) { unsent = undefined; return; } throw replacementError; } + } + // An earlier claim may have been reclaimed after the initial snapshot, even + // when insertOne succeeded. Persist the current fence before resource preparation. + const current = await collection.findOne({ ...identity, accessAlertGeneration: generationFilter(record.generation) }, + { projection: { accessAlertClaimRevision: 1 } }); + if (!current) return; + const revision = current.accessAlertClaimRevision ?? 0; + const refreshed = await alerts.updateOne({ _id: key, attemptId, state: "pending", + resetRevision: generationFilter(record.resetRevision) }, { $set: { resourceRevision: revision } }); + if (!refreshed.matchedCount) return; + record.resourceRevision = revision; 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() }); + const claim = await collection.updateOne({ ...identity, + accessAlertGeneration: generationFilter(record.generation), + accessAlertClaimRevision: revisionFilter(record.resourceRevision ?? 0), + }, { $set: { accessAlertClaimedAt: new Date(), accessAlertAttemptId: attemptId }, $inc: { accessAlertClaimRevision: 1 } }); + if (!claim.matchedCount || !await collection.findOne({ ...identity, + accessAlertGeneration: generationFilter(record.generation), + accessAlertClaimRevision: (record.resourceRevision ?? 0) + 1, accessAlertAttemptId: attemptId, + })) return; + // Reserve the recipient transition, but it remains reclaimable until the owner CAS commits dispatch. + const dispatch = await alerts.updateOne({ _id: key, attemptId, state: "pending", + resetRevision: generationFilter(record.resetRevision) }, { $set: { state: "sending" } }); + if (!dispatch.matchedCount) return; + // Reset and dispatch commit on the same owner document. There is no asynchronous + // preparation between this authoritative commitment and the provider request. + if (!await commitOwnerRevision(identity.owner, ownerRevision, record)) return; + let accepted = false; + let httpStatus: number | undefined; try { + providerStarted = true; const response = await fetch("https://api.resend.com/emails", { method: "POST", - headers: { Authorization: `Bearer ${config.RESEND_API_KEY}`, "Content-Type": "application/json" }, + headers: { Authorization: `Bearer ${config.RESEND_API_KEY}`, "Content-Type": "application/json", + "Idempotency-Key": `owner-access-alert/${attemptId}` }, signal: AbortSignal.timeout(15000), body: JSON.stringify({ from: config.EMAIL_FROM, to: [to], @@ -97,14 +547,35 @@ export async function notifyOwnerAccessProblem(ownerId: string, error: unknown, text: `Anonymous GitHub could not access GitHub using your connection, or could not read a repository.\n\nOpen your connections page to reconnect your account or restore repository access:\n${url}\n\nYou are receiving this email because you own content on Anonymous GitHub. You can change your alert address on the connections page.\n`, }), }); - if (!response.ok) throw new Error("email_delivery_failed"); - } catch { - // Keep the claim even on failure: delivery can be ambiguous and no - // automatic retry may send another email before owner action. - logger.warn("owner access email delivery failed"); - } + httpStatus = response.status; + accepted = response.ok; + } catch { /* Delivery can be ambiguous; retain the recipient guard. */ } + const state = accepted ? "accepted" : "failed"; + if (accepted) logger.info("owner access email accepted", { attemptId, httpStatus }); + else logger.warn("owner access email delivery failed", { attemptId, httpStatus }); + const completedAt = new Date(); + try { + await retryAlertWrite(async () => { + try { await alertCollection("owner_access_alert_events").insertOne({ _id: `${attemptId}/completion`, attemptId, + recipientKey: key, state, httpStatus, completedAt }); } + catch (error) { if ((error as { code?: number }).code !== 11000) throw error; } + }); + } catch { logger.warn("owner access email audit unavailable", { attemptId }); } + try { + await retryAlertWrite(async () => { + await alerts.updateOne({ _id: key, attemptId, state: "sending" }, { $set: { state, completedAt, httpStatus } }); + await clearOwnerDispatch(record, "committed"); + await applyDeferredRecoveries(key, attemptId); + }); + } catch { logger.warn("owner access email result reconciliation required", { attemptId }); } } catch { - // Do not log provider errors, which can contain credentials or addresses. logger.warn("owner access email unavailable"); + } finally { + if (unsent && !providerStarted) { + await releaseUnsentAlert(unsent, { state: { $in: ["pending", "sending", "cancelled"] } }) + .catch(() => logger.warn("owner access email claim cleanup unavailable")); + } + if (dispatchRecord) await clearOwnerDispatch(dispatchRecord, "prepared") + .catch(() => logger.warn("owner access email preparation cleanup unavailable")); } } diff --git a/src/core/source/GitHubStream.ts b/src/core/source/GitHubStream.ts index 7b92778..564b7aa 100644 --- a/src/core/source/GitHubStream.ts +++ b/src/core/source/GitHubStream.ts @@ -1,3 +1,4 @@ +import { withOwnerAccessRecovery } from "../owner-notifications"; import { ExpiringMap } from "../expiring-map"; import { contentGenerationPrefix, contentRetirementMarker } from "../content-generation"; import { createHash } from "crypto"; @@ -391,7 +392,7 @@ export default class GitHubStream extends GitHubBase { const missed = missingContent.get(missKey); if (missed) throw new AnonymousError(missed, { httpStatus: 404 }); try { - return await measureStage("upstream", async () => { + const fetchContent = () => measureStage("upstream", async () => { const content = await this.downloadWithFallback(token, expected.sha, filePath) .catch(error => { throw downloadError(error, filePath); }); let count = 0; @@ -412,6 +413,7 @@ export default class GitHubStream extends GitHubBase { finally { clearTimeout(deadline); limited.destroy(); content.destroy(); } return true; }); + return await this.confirmOwnerAccess(repoId, fetchContent); } catch (error) { if (error instanceof AnonymousError && error.message === "file_not_found" && error.httpStatus === 404) { const code = await classifyContentMiss(this.data, token); @@ -464,6 +466,25 @@ export default class GitHubStream extends GitHubBase { * files above the 1MB contents-API inline limit. */ async fetchFileInfoFromPath(filePath: string): Promise { + return this.confirmOwnerAccess(this.data.repoId, () => this.fetchFileMetadata(filePath), result => result !== null); + } + + private async confirmOwnerAccess(repoId: string, work: () => Promise, confirmed?: (result: T) => boolean): Promise { + let repository; + try { + repository = isConnected ? await AnonymizedRepositoryModel.findOne({ repoId }) + .select("owner source githubAccess").lean().exec() : null; + } catch { + // Email recovery bookkeeping must not prevent the upstream read. + return work(); + } + return repository?.owner ? withOwnerAccessRecovery(String(repository.owner), + { kind: "repository", id: String(repository._id) }, work, + { "source.repositoryName": `${this.data.organization}/${this.data.repoName}`, "source.commit": this.data.commit, + "githubAccess.revision": repository.githubAccess?.revision ?? { $exists: false } }, confirmed) : work(); + } + + private async fetchFileMetadata(filePath: string): Promise { const token = await this.data.getToken(); const oct = octokit(token); try { diff --git a/src/scripts/reconcile-legacy-owner-alerts.ts b/src/scripts/reconcile-legacy-owner-alerts.ts new file mode 100644 index 0000000..070cf55 --- /dev/null +++ b/src/scripts/reconcile-legacy-owner-alerts.ts @@ -0,0 +1,17 @@ +import mongoose from "mongoose"; +import config from "../config"; +import { getMongoUrl } from "../server/database"; +import { prepareLegacyOwnerAlertMigration } from "../core/owner-notifications"; + +async function main(): Promise { + await mongoose.connect(getMongoUrl(), config.MONGODB_URI ? {} : { authSource: "admin" }); + try { + await prepareLegacyOwnerAlertMigration(); + process.stdout.write("Legacy owner alert reconciliation prepared\n"); + } finally { await mongoose.disconnect(); } +} + +void main().catch(() => { + process.stderr.write("Could not prepare legacy owner alert reconciliation\n"); + process.exitCode = 1; +}); diff --git a/src/server/database.ts b/src/server/database.ts index 4f6196e..fb4457f 100644 --- a/src/server/database.ts +++ b/src/server/database.ts @@ -4,6 +4,7 @@ import CredentialModel from "../core/model/credentials/credentials.model"; import mongoose, { ConnectOptions } from "mongoose"; import Repository from "../core/Repository"; import config from "../config"; +import { initializeOwnerAccessAlerts } from "../core/owner-notifications"; import AnonymizedRepositoryModel from "../core/model/anonymizedRepositories/anonymizedRepositories.model"; import AnonymousError from "../core/AnonymousError"; import AnonymizedPullRequestModel from "../core/model/anonymizedPullRequests/anonymizedPullRequests.model"; @@ -34,6 +35,7 @@ export async function connect(overrides: Pick { if (isDisabledAccount(user.status)) throw appError("not_connected", 403); await saveAppGrant(user.id, tokens); await saveRegistrationEmail(user, tokens.access_token); - await resetOwnerAccessAlerts(user.id); + if (flow.ownerId) await resetOwnerAccessAlerts(user.id); await new Promise((resolve, reject) => req.login({ username: user!.username, user }, err => err ? reject(err) : resolve())); res.redirect(flow.install ? `/github/app/install?returnTo=${encodeURIComponent(flow.returnTo)}&repository=${encodeURIComponent(flow.repository || "")}` : flow.returnTo); } catch (error) { handleError(error, res, req); } diff --git a/test/deploy-owner-alert-migration.test.js b/test/deploy-owner-alert-migration.test.js new file mode 100644 index 0000000..9a737f5 --- /dev/null +++ b/test/deploy-owner-alert-migration.test.js @@ -0,0 +1,101 @@ +const { expect } = require("chai"); +const { spawnSync } = require("child_process"); +const fs = require("fs"); +const os = require("os"); +const path = require("path"); + +describe("owner alert migration deployment", function () { + this.timeout(10000); + function deploy(options = {}) { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), "owner-alert-deploy-")); + const statePath = path.join(directory, "state.json"), logPath = path.join(directory, "calls.jsonl"); + fs.mkdirSync(path.join(directory, "scripts")); + fs.copyFileSync(path.join(__dirname, "../scripts/deploy.sh"), path.join(directory, "scripts/deploy.sh")); + fs.writeFileSync(statePath, JSON.stringify({ ...options })); + fs.writeFileSync(path.join(directory, "docker"), `#!/usr/bin/env node +const fs = require("fs"); +const args = process.argv.slice(2), state = JSON.parse(fs.readFileSync(process.env.TASK_DOCKER_STATE)); +fs.appendFileSync(process.env.TASK_DOCKER_LOG, JSON.stringify(args) + "\\n"); +const finish = (output = "", code = 0) => { fs.writeFileSync(process.env.TASK_DOCKER_STATE, JSON.stringify(state)); process.stdout.write(output); process.exit(code); }; +if (args[0] === "image" && args[1] === "inspect") { + if (args.some(a => a.includes("Labels"))) finish(state.badProtocol ? "" : "recipient-v1"); + finish(state.built ? "sha:new" : "sha:old"); +} +if (args[0] === "inspect") { + if (args.some(a => a.includes("Labels"))) finish(""); + if (args.includes("{{.State.Running}}")) finish(state.stopped && !state.refuseStop ? "false" : "true"); + finish("healthy"); +} +if (args[0] === "compose") { + if (args[1] === "build") { if (state.buildFails) finish("", 1); state.built = true; finish(); } + if (args[1] === "stop") { state.stopped = true; finish(); } + if (args[1] === "run") { if (state.reconcileFails) finish("", 1); state.reconciled = true; finish(); } + if (args[1] === "up") { if (args.includes("streamer")) state.streamers = true; else state.api = true; finish(); } + if (args[1] === "ps") { + if (args.includes("-q")) finish(args.includes("streamer") ? "old-api\\nold-1\\nold-2\\nold-3\\nold-4\\n" : "new-api"); + if (args.includes("streamer")) finish(state.streamers ? "new-1\\nnew-2\\nnew-3\\nnew-4\\n" : "old-1\\nold-2\\nold-3\\nold-4\\n"); + finish(); + } +} +finish("unexpected docker call", 2); +`, { mode: 0o755 }); + try { + const result = spawnSync("bash", [path.join(directory, "scripts/deploy.sh")], { + encoding: "utf8", timeout: 8000, + env: { ...process.env, PATH: directory + path.delimiter + process.env.PATH, + FORCE: String(options.force ?? 1), STREAMER_REPLICAS: "4", TASK_DOCKER_STATE: statePath, TASK_DOCKER_LOG: logPath }, + }); + return { ...result, calls: fs.readFileSync(logPath, "utf8").trim().split("\n").map(line => JSON.parse(line)) }; + } finally { fs.rmSync(directory, { recursive: true, force: true }); } + } + + it("stops and verifies every legacy writer before starting any migration-capable instance", () => { + const result = deploy(); + expect(result.status, result.stderr).to.equal(0); + const stop = result.calls.findIndex(args => args[0] === "compose" && args[1] === "stop"); + const up = result.calls.findIndex(args => args[0] === "compose" && args[1] === "up"); + expect(stop).to.be.greaterThan(-1); expect(up).to.be.greaterThan(stop); + const reconcile = result.calls.findIndex(args => args[0] === "compose" && args[1] === "run"); + expect(reconcile).to.be.greaterThan(stop); expect(reconcile).to.be.lessThan(up); + expect(result.calls[reconcile]).to.include("./build/scripts/reconcile-legacy-owner-alerts.js"); + const checked = result.calls.slice(stop + 1, up).filter(args => args[0] === "inspect").map(args => args.at(-1)); + expect(checked).to.deep.equal(["old-api", "old-1", "old-2", "old-3", "old-4"]); + expect(result.calls[stop]).to.include.members(["anonymous_github", "streamer"]); + expect(result.calls[up]).to.include.members(["--force-recreate", "streamer=4"]); + const api = result.calls.findIndex(args => args[0] === "compose" && args[1] === "up" && args.at(-1) === "anonymous_github"); + expect(api).to.be.greaterThan(up); + for (const name of ["new-1", "new-2", "new-3", "new-4"]) { + expect(result.calls.slice(up + 1, api).some(args => args[0] === "inspect" && args.at(-1) === name)).to.equal(true); + } + }); + it("aborts before migration starts if a legacy writer remains running", () => { + const result = deploy({ refuseStop: true }); + expect(result.status).not.to.equal(0); + expect(result.stderr).to.include("still running"); + expect(result.calls.some(args => args[0] === "compose" && args[1] === "up")).to.equal(false); + }); + it("keeps old containers running if the image build fails", () => { + const result = deploy({ buildFails: true }); + expect(result.status).not.to.equal(0); + expect(result.calls.some(args => args[0] === "compose" && ["stop", "up"].includes(args[1]))).to.equal(false); + }); + it("keeps old containers running if the replacement lacks migration support", () => { + const result = deploy({ badProtocol: true }); + expect(result.status).not.to.equal(0); + expect(result.stderr).to.include("lacks recipient alert migration support"); + expect(result.calls.some(args => args[0] === "compose" && ["stop", "up"].includes(args[1]))).to.equal(false); + }); + it("starts no application instance if rollback reconciliation cannot be prepared", () => { + const result = deploy({ reconcileFails: true }); + expect(result.status).not.to.equal(0); + expect(result.calls.some(args => args[0] === "compose" && args[1] === "up")).to.equal(false); + }); + for (const stopped of [true, false]) it(`reconciles legacy ${stopped ? "stopped" : "running"} containers when the replacement was already built`, () => { + const result = deploy({ built: true, stopped, force: 0 }); + expect(result.status, result.stderr).to.equal(0); + const reconcile = result.calls.findIndex(args => args[0] === "compose" && args[1] === "run"); + const up = result.calls.findIndex(args => args[0] === "compose" && args[1] === "up"); + expect(reconcile).to.be.greaterThan(-1); expect(up).to.be.greaterThan(reconcile); + expect(result.stdout).not.to.include("nothing to deploy"); + }); +}); diff --git a/test/owner-notifications.test.js b/test/owner-notifications.test.js index 4b0738a..2c42bc4 100644 --- a/test/owner-notifications.test.js +++ b/test/owner-notifications.test.js @@ -11,37 +11,122 @@ describe("owner access notifications", () => { const ownerId = "507f1f77bcf86cd799439011"; const resource = { kind: "repository", id: "507f1f77bcf86cd799439012" }; const other = { kind: "repository", id: "507f1f77bcf86cd799439013" }; - let owner, sent, claims, rows, generations; + let owner, ownerState, sent, claims, rows, generations, resourceRevisions, attemptIds, recipients, events; beforeEach(() => { owner = { status: "active", emails: [{ email: "owner@example.com", default: true }] }; - sent = []; claims = []; rows = new Map(); generations = new Map(); + ownerState = {}; + sent = []; claims = []; rows = new Map(); generations = new Map(); recipients = new Map(); events = []; + resourceRevisions = new Map(); attemptIds = new Map(); stub(config, "RESEND_API_KEY", "test-resend-key"); stub(config, "EMAIL_FROM", "alerts@example.com"); stub(config, "APP_HOSTNAME", "anonymous.example.com"); stub(User, "findOne", () => ({ lean: async () => ["removed", "banned"].includes(owner?.status) ? null : owner })); + stub(User, "findById", () => ({ select: () => ({ lean: async () => owner }) })); Object.defineProperty(User.db, "readyState", { configurable: true, value: 1 }); restores.push(() => { delete User.db.readyState; }); - stub(User.db, "collection", name => ({ - findOne: async filter => { - const key = `${name}:${filter.owner}:${filter._id}`; - if (!generations.has(key)) generations.set(key, undefined); - return { accessAlertGeneration: generations.get(key) }; - }, - updateOne: async (filter, update) => { - claims.push(filter); - const key = `${name}:${filter.owner}:${filter._id}`; - if (rows.has(key) || generations.get(key) !== (typeof filter.accessAlertGeneration === "string" ? filter.accessAlertGeneration : undefined)) return { modifiedCount: 0 }; - rows.set(key, update.$set.accessAlertClaimedAt); - return { modifiedCount: 1 }; - }, - updateMany: async (filter, update) => { - for (const key of generations.keys()) if (key.startsWith(`${name}:${filter.owner}:`) && (!filter._id || key.endsWith(`:${filter._id}`))) { - rows.delete(key); - generations.set(key, update.$set.accessAlertGeneration); - } - }, - })); - stub(global, "fetch", async (url, options) => { sent.push({ url, options, body: JSON.parse(options.body) }); return { ok: true }; }); + stub(User.db, "collection", name => { + if (name === "users") return { + findOne: async filter => String(filter._id) === ownerId ? { ...ownerState } : null, + findOneAndUpdate: async (filter, update) => { + if (String(filter._id) !== ownerId) return { value: null }; + ownerState.accessAlertResetRevision = (ownerState.accessAlertResetRevision || 0) + update.$inc.accessAlertResetRevision; + Object.assign(ownerState, update.$set); return { value: { ...ownerState } }; + }, + updateOne: async (filter, update) => { + if (String(filter._id) !== ownerId) return { matchedCount: 0 }; + if ("accessAlertResetRevision" in filter && ownerState.accessAlertResetRevision !== (typeof filter.accessAlertResetRevision === "number" ? filter.accessAlertResetRevision : undefined)) return { matchedCount: 0 }; + if (filter.$or && ownerState.accessAlertResetUntil?.getTime() > Date.now()) return { matchedCount: 0 }; + const get = path => path.split(".").reduce((value, key) => value?.[key], ownerState); + const set = (path, value) => { + const keys = path.split("."), last = keys.pop(); let target = ownerState; + for (const key of keys) target = target[key] ||= {}; + if (value === undefined) delete target[last]; else target[last] = value; + }; + for (const [key, value] of Object.entries(filter)) if (key.startsWith("accessAlertDispatches.") && get(key) !== value) return { matchedCount: 0 }; + if (Array.isArray(update)) { + const cutoff = update[0].$set.accessAlertDispatches.$arrayToObject.$filter.cond.$or[1].$gt[1]; + ownerState.accessAlertDispatches = Object.fromEntries(Object.entries(ownerState.accessAlertDispatches || {}).filter(([, value]) => value.state === "committed" || value.claimedAt > cutoff)); + for (const [key, value] of Object.entries(update[1].$set)) set(key, value); + } else { + if (update.$inc) for (const [key, value] of Object.entries(update.$inc)) set(key, (get(key) || 0) + value); + if (update.$set) for (const [key, value] of Object.entries(update.$set)) set(key, value); + if (update.$unset) for (const key of Object.keys(update.$unset)) set(key, undefined); + } + return { matchedCount: 1 }; + }, + }; + if (name === "owner_access_alert_events") return { insertOne: async record => events.push(record), findOne: async filter => events.find(event => event._id === filter._id) }; + if (name === notifications.RECIPIENT_ALERTS) return { + findOne: async filter => recipients.get(filter._id), + find: filter => { + const matching = () => [...recipients.values()].filter(r => { + if (filter.deferredRecoveries) return r.deferredRecoveries?.some(recovery => String(recovery.owner) === String(filter.deferredRecoveries.$elemMatch.owner)); + const claim = filter.$or[0]; + return String(r.owner) === String(claim.owner) && r.resourceKind === claim.resourceKind + && String(r.resourceId) === String(claim.resourceId) && r.generation !== claim.generation.$ne && filter.state.$in.includes(r.state); + }).map(r => ({ ...r })); + return { toArray: async () => matching(), async *[Symbol.asyncIterator]() { yield* matching(); } }; + }, + updateMany: async filter => { if (filter.legacyClaims?.$elemMatch) return; const claim = filter; for (const r of recipients.values()) if (String(r.owner) === String(claim.owner) + && (!claim.resourceId || String(r.resourceId) === String(claim.resourceId))) r.resetRevision = (r.resetRevision || 0) + 1; }, + insertOne: async record => { if (recipients.has(record._id)) throw Object.assign(new Error("duplicate"), { code: 11000 }); recipients.set(record._id, { ...record }); }, + updateOne: async (filter, update) => { const record = recipients.get(filter._id); + const stateMatches = !filter.state || (typeof filter.state === "string" ? record?.state === filter.state : filter.state.$in.includes(record?.state)); + const revisionMatches = !("resetRevision" in filter) || record?.resetRevision === (typeof filter.resetRevision === "number" ? filter.resetRevision : undefined); + if (!record || record.attemptId !== filter.attemptId || !stateMatches || !revisionMatches) return { matchedCount: 0 }; + if (Array.isArray(update)) { + const recovery = update[0].$set.deferredRecoveries.$concatArrays[1][0]; + record.deferredRecoveries = [recovery]; + } else Object.assign(record, update.$set); + return { matchedCount: 1 }; + }, + deleteOne: async filter => { const r = recipients.get(filter._id); + const stateMatches = typeof filter.state === "string" ? r?.state === filter.state : filter.state.$in.includes(r?.state); + const revisionMatches = !("resetRevision" in filter) || r?.resetRevision === (typeof filter.resetRevision === "number" ? filter.resetRevision : undefined); + if (r?.attemptId === filter.attemptId && stateMatches && revisionMatches) { recipients.delete(filter._id); return { deletedCount: 1 }; } + return { deletedCount: 0 }; + }, + deleteMany: async filter => { let deletedCount = 0; for (const [key, r] of recipients) { + if (String(r.owner) === String(filter.owner) && r.resourceKind === filter.resourceKind && String(r.resourceId) === String(filter.resourceId) + && r.generation !== filter.generation.$ne && filter.state.$in.includes(r.state)) { recipients.delete(key); deletedCount++; } + } return { deletedCount }; }, + }; + return { + findOne: async filter => { + const key = `${name}:${filter.owner}:${filter._id}`; + if (!generations.has(key)) generations.set(key, undefined); + if ("accessAlertGeneration" in filter && generations.get(key) !== (typeof filter.accessAlertGeneration === "string" ? filter.accessAlertGeneration : undefined)) return null; + if (filter.accessAlertClaimedAt?.$exists && !rows.has(key)) return null; + if ("accessAlertClaimRevision" in filter && (resourceRevisions.get(key) || 0) !== (typeof filter.accessAlertClaimRevision === "number" ? filter.accessAlertClaimRevision : 0)) return null; + if (filter.accessAlertAttemptId && attemptIds.get(key) !== filter.accessAlertAttemptId) return null; + return { accessAlertGeneration: generations.get(key), accessAlertClaimedAt: rows.get(key), accessAlertClaimRevision: resourceRevisions.get(key), status: "ready" }; + }, + updateOne: async (filter, update) => { + claims.push(filter); const key = `${name}:${filter.owner}:${filter._id}`; + if (generations.get(key) !== (typeof filter.accessAlertGeneration === "string" ? filter.accessAlertGeneration : undefined)) return { matchedCount: 0 }; + const revision = resourceRevisions.get(key) || 0; + if (Array.isArray(update)) { + const first = filter.$or[0].accessAlertClaimRevision; + const expected = typeof first === "number" ? first : 0; + const attemptId = update[0].$set.accessAlertAttemptId.$cond[0].$eq[1]; + if (revision !== expected && !(revision === expected + 1 && attemptIds.get(key) === attemptId)) return { matchedCount: 0 }; + resourceRevisions.set(key, revision + 1); + if (attemptIds.get(key) === attemptId) { rows.delete(key); attemptIds.delete(key); } + } else { + if ("accessAlertClaimRevision" in filter && revision !== (typeof filter.accessAlertClaimRevision === "number" ? filter.accessAlertClaimRevision : 0)) return { matchedCount: 0 }; + rows.set(key, update.$set.accessAlertClaimedAt); attemptIds.set(key, update.$set.accessAlertAttemptId); + resourceRevisions.set(key, revision + update.$inc.accessAlertClaimRevision); + } + return { matchedCount: 1 }; + }, + updateMany: async (filter, update) => { + for (const key of generations.keys()) if (key.startsWith(`${name}:${filter.owner}:`) && (!filter._id || key.endsWith(`:${filter._id}`))) { + rows.delete(key); attemptIds.delete(key); generations.set(key, update.$set.accessAlertGeneration); + } + }, + }; + }); + stub(global, "fetch", async (url, options) => { sent.push({ url, options, body: JSON.parse(options.body) }); return { ok: true, status: 200 }; }); }); afterEach(() => { while (restores.length) restores.pop()(); }); @@ -53,30 +138,55 @@ describe("owner access notifications", () => { expect(notificationEmail([{ email: "first@example.com", default: false }, { email: "primary@example.com", default: true }])).to.equal("primary@example.com"); }); - it("sends once per resource across concurrent readers and repeated errors, with no time expiry", async () => { + it("sends once per recipient across resources and concurrent readers, with no time expiry", async () => { await Promise.all(Array.from({ length: 10 }, () => notifyOwnerAccessProblem(ownerId, "token_expired", resource))); expect(sent).to.have.length(1); for (const key of rows.keys()) rows.set(key, new Date("2000-01-01")); await notifyOwnerAccessProblem(ownerId, "repo_not_found", resource); expect(sent).to.have.length(1); await notifyOwnerAccessProblem(ownerId, "repo_not_found", other); - expect(sent).to.have.length(2); + expect(sent).to.have.length(1); expect(sent[0].url).to.equal("https://api.resend.com/emails"); expect(sent[0].body.to).to.deep.equal(["owner@example.com"]); expect(sent[0].body.text).to.include("https://anonymous.example.com/connections"); - expect(claims[0].accessAlertClaimedAt).to.deep.equal({ $exists: false }); + expect(sent[0].options.headers["Idempotency-Key"]).to.match(/^owner-access-alert\//); + expect(events[0].state).to.equal("prepared"); + expect(events[1].state).to.equal("accepted"); }); - it("owner action resets only the selected resource and cannot reset another owner's alerts", async () => { + it("reopens an alert only after owner action and successful access to its triggering resource", async () => { await notifyOwnerAccessProblem(ownerId, "token_expired", resource); await notifyOwnerAccessProblem(ownerId, "token_expired", other); - await notifications.resetOwnerAccessAlerts("507f1f77bcf86cd799439099", resource); - await notifyOwnerAccessProblem(ownerId, "token_expired", resource); - expect(sent).to.have.length(2); + expect(sent).to.have.length(1); await notifications.resetOwnerAccessAlerts(ownerId, resource); await notifyOwnerAccessProblem(ownerId, "token_expired", resource); - await notifyOwnerAccessProblem(ownerId, "token_expired", other); - expect(sent).to.have.length(3); + expect(sent).to.have.length(1); + await notifications.withOwnerAccessRecovery(ownerId, other, async () => {}); + expect(recipients.size).to.equal(1); + await notifications.withOwnerAccessRecovery(ownerId, resource, async () => {}); + await notifyOwnerAccessProblem(ownerId, "token_expired", resource); + expect(sent).to.have.length(2); + expect(events).to.have.length(4); + }); + + it("cannot clear another owner's recipient guard", async () => { + await notifyOwnerAccessProblem(ownerId, "token_expired", resource); + await notifications.resetOwnerAccessAlerts("507f1f77bcf86cd799439099", resource); + await notifications.withOwnerAccessRecovery("507f1f77bcf86cd799439099", resource, async () => {}); + expect(recipients.size).to.equal(1); + }); + + it("does not reopen on background success without owner action", async () => { + await notifyOwnerAccessProblem(ownerId, "token_expired", resource); + await notifications.withOwnerAccessRecovery(ownerId, resource, async () => {}); + expect(recipients.size).to.equal(1); + }); + + it("retains the guard when access recovery fails", async () => { + await notifyOwnerAccessProblem(ownerId, "token_expired", resource); + await notifications.resetOwnerAccessAlerts(ownerId, resource); + try { await notifications.withOwnerAccessRecovery(ownerId, resource, async () => { throw new Error("still forbidden"); }); } catch { /* Expected upstream failure. */ } + expect(recipients.size).to.equal(1); }); it("rejects a pre-action failure paused during owner lookup", async () => { @@ -98,16 +208,16 @@ describe("owner access notifications", () => { expect(sent).to.have.length(1); }); - it("reconnecting resets all of the owner's resource types", async () => { + it("reconnecting leaves the recipient blocked across all resource types", async () => { const resources = [resource, { ...resource, kind: "pull-request" }, { ...resource, kind: "gist" }]; for (const item of resources) await notifyOwnerAccessProblem(ownerId, "token_expired", item); - expect(sent).to.have.length(3); + expect(sent).to.have.length(1); await notifications.resetOwnerAccessAlerts(ownerId); for (const item of resources) await notifyOwnerAccessProblem(ownerId, "token_expired", item); - expect(sent).to.have.length(6); + expect(sent).to.have.length(1); }); - for (const failure of ["rejected", "timeout"]) it(`retains the claim after ${failure} until owner action`, async () => { + for (const failure of ["rejected", "timeout"]) it(`retains the claim after ${failure} until confirmed recovery`, async () => { let attempts = 0; stub(global, "fetch", async () => { attempts++; if (failure === "timeout") throw new Error("timeout"); return { ok: false }; }); await notifyOwnerAccessProblem(ownerId, "token_expired", resource); @@ -115,6 +225,9 @@ describe("owner access notifications", () => { expect(attempts).to.equal(1); await notifications.resetOwnerAccessAlerts(ownerId, resource); await notifyOwnerAccessProblem(ownerId, "token_expired", resource); + expect(attempts).to.equal(1); + await notifications.withOwnerAccessRecovery(ownerId, resource, async () => {}); + await notifyOwnerAccessProblem(ownerId, "token_expired", resource); expect(attempts).to.equal(2); }); @@ -423,12 +536,54 @@ describe("repository probes and truncated file recovery", () => { const Files = require("../src/core/model/files/files.model").default; const File = require("../src/core/AnonymizedFile").default; const GitHubStream = require("../src/core/source/GitHubStream").default; + const AnonymizedRepositories = require("../src/core/model/anonymizedRepositories/anonymizedRepositories.model").default; + const storage = require("../src/core/storage").default; + const { Readable } = require("node:stream"); const { classifyGitHubMissError } = require("../src/core/source/GitHubBase"); const restores = []; const stub = (object, key, value) => { const old = object[key]; restores.push(() => object[key] = old); object[key] = value; }; const source = { getToken: async () => "probe-test", repoId: "anon", organization: "owner", repoName: "repo", commit: "sha" }; const missing = () => Object.assign(new Error("Not found"), { status: 404 }); afterEach(() => { while (restores.length) restores.pop()(); }); + for (const failed of [false, true]) it(`continues truncated metadata reads after recovery lookup failure: ${failed ? "upstream failure" : "success"}`, async () => { + const upstreamError = Object.assign(new Error("GitHub unavailable"), { status: 503 }); + let reads = 0; + stub(db, "isConnected", true); + stub(AnonymizedRepositories, "findOne", () => ({ select: () => ({ lean: () => ({ exec: async () => { throw new Error("recovery database unavailable"); } }) }) })); + stub(notifications, "withOwnerAccessRecovery", () => { throw new Error("recovery must be skipped without metadata"); }); + stub(github, "octokit", () => ({ repos: { getContent: async () => { + reads++; if (failed) throw upstreamError; + return { data: { type: "file", sha: "blob", size: 11 } }; + } } })); + const result = await new GitHubStream(source).fetchFileInfoFromPath("large/file.txt").then(value => ({ value }), error => ({ error })); + expect(reads).to.equal(1); + if (failed) expect(result.error).to.equal(upstreamError); + else expect(result.value).to.include({ name: "file.txt", path: "large", sha: "blob", size: 11 }); + }); + for (const failed of [false, true]) it(`continues uncached file reads after recovery lookup failure: ${failed ? "upstream failure" : "success"}`, async () => { + const upstreamError = Object.assign(new Error("GitHub unavailable"), { status: 503 }); + const data = { ...source, repoId: `lookup-failure-${failed}` }; + let cached, downloads = 0, recoveryLookups = 0; + stub(db, "isConnected", true); + stub(AnonymizedRepositories, "findOne", () => ({ select: fields => ({ lean: () => ({ exec: async () => { + if (fields === "owner source githubAccess") { recoveryLookups++; throw new Error("recovery database unavailable"); } + return { status: "ready" }; + } }) }) })); + stub(notifications, "withOwnerAccessRecovery", () => { throw new Error("recovery must be skipped without metadata"); }); + stub(storage, "fileInfo", async () => { if (!cached) throw Object.assign(new Error("missing"), { code: "ENOENT" }); return { size: cached.length }; }); + stub(storage, "write", async (_repo, _path, input) => { const chunks = []; for await (const chunk of input) chunks.push(chunk); cached = Buffer.concat(chunks); }); + stub(storage, "read", async () => Readable.from([cached])); + const stream = new GitHubStream(data); + stream.downloadWithFallback = async () => { downloads++; if (failed) throw upstreamError; return Readable.from([Buffer.from("fresh-bytes")]); }; + const result = await stream.getFileContentCache("file.txt", data.repoId, () => ({ sha: "blob", size: 11 })) + .then(value => ({ value }), error => ({ error })); + expect(downloads).to.equal(1); expect(recoveryLookups).to.equal(1); + if (failed) { expect(result.error.message).to.equal("upstream_error"); expect(result.error.cause).to.equal(upstreamError); } + else { + const chunks = []; for await (const chunk of result.value) chunks.push(chunk); + expect(Buffer.concat(chunks).toString()).to.equal("fresh-bytes"); + } + }); for (const stage of ["name", "id"]) for (const status of [undefined, 429, 503]) { it(`propagates inconclusive ${stage} probe ${status}`, async () => { const error = Object.assign(new Error("probe failed"), { status }); diff --git a/test/recipient-alerts-mongo.test.js b/test/recipient-alerts-mongo.test.js new file mode 100644 index 0000000..242579e --- /dev/null +++ b/test/recipient-alerts-mongo.test.js @@ -0,0 +1,742 @@ +const { expect } = require("chai"); +const mongoose = require("mongoose"); +require("ts-node/register/transpile-only"); +const Users = require("../src/core/model/users/users.model").default; +const config = require("../src/config").default; +const alerts = require("../src/core/owner-notifications"); +const { notifyOwnerAccessProblem: notify, resetOwnerAccessAlerts: reset, withOwnerAccessRecovery: recover } = alerts; +const describeMongo = process.env.TEST_MONGODB_URI ? describe : describe.skip; + +describeMongo("recipient access alerts in MongoDB", function () { + this.timeout(15000); + let owner, otherOwner, resource, other, sent, nativeFetch, originalConfig; + const coll = kind => mongoose.connection.collection({ repository: "anonymizedrepositories", gist: "anonymizedgists", "pull-request": "anonymizedpullrequests" }[kind]); + const guards = () => mongoose.connection.collection(alerts.RECIPIENT_ALERTS); + const recipient = () => alerts.recipientAlertKey("owner@example.com"); + async function create(ownerId, kind = "repository") { + const id = new mongoose.Types.ObjectId(); await coll(kind).insertOne({ _id: id, owner: ownerId, repoId: String(id), status: "ready" }); return { kind, id: String(id) }; + } + before(async () => { + nativeFetch = global.fetch; originalConfig = { ...config }; + await mongoose.connect(process.env.TEST_MONGODB_URI, { dbName: "recipient_alerts_" + Date.now() }); + Object.assign(config, { RESEND_API_KEY: "test-key", EMAIL_FROM: "alerts@example.com", APP_HOSTNAME: "anonymous.example.com" }); + }); + after(async () => { global.fetch = nativeFetch; Object.assign(config, originalConfig); await mongoose.connection.dropDatabase(); await mongoose.disconnect(); }); + beforeEach(async () => { + for (const c of Object.values(mongoose.connection.collections)) await c.deleteMany({}); + owner = await Users.create({ username: "owner", emails: [{ email: "owner@example.com", default: true }] }); + otherOwner = await Users.create({ username: "other-owner", notificationEmail: "OWNER@example.com" }); + resource = await create(owner._id); other = await create(otherOwner._id, "gist"); sent = []; + global.fetch = async (_url, options) => { sent.push(options); return { ok: true, status: 200 }; }; + }); + + it("allows only one sender across owners, resources and case variations under concurrent failures", async () => { + await Promise.all(Array.from({ length: 40 }, (_, i) => notify(i % 2 ? owner.id : otherOwner.id, "token_expired", i % 2 ? resource : other))); + expect(sent).to.have.length(1); expect(await guards().countDocuments({ _id: recipient() })).to.equal(1); + expect(await mongoose.connection.collection("owner_access_alert_events").countDocuments({ state: "accepted" })).to.equal(1); + }); + it("does not reopen after reconnect alone, unrelated resource success, or failed recovery", async () => { + await notify(owner.id, "token_expired", resource); await reset(owner.id); + await recover(otherOwner.id, other, async () => {}); + try { await recover(owner.id, resource, async () => { throw new Error("Forbidden"); }); } catch { /* Expected. */ } + await notify(owner.id, "token_expired", resource); await notify(otherOwner.id, "token_expired", other); + expect(sent).to.have.length(1); + }); + it("requires owner action before a successful upstream read can reopen the recipient", async () => { + await notify(owner.id, "token_expired", resource); + await recover(owner.id, resource, async () => {}); expect(await guards().findOne({ _id: recipient() })).not.to.equal(null); + await reset(owner.id, resource); await recover(owner.id, resource, async () => {}); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + await notify(owner.id, "token_expired", resource); expect(sent).to.have.length(2); + expect(sent[0].headers["Idempotency-Key"]).not.to.equal(sent[1].headers["Idempotency-Key"]); + expect(await mongoose.connection.collection("owner_access_alert_events").countDocuments()).to.equal(4); + }); + it("does not release a pending send when recovery races with provider delivery", async () => { + let release, entered; const gate = new Promise(resolve => { entered = resolve; }); + global.fetch = async (_url, options) => { sent.push(options); entered(); await new Promise(resolve => { release = resolve; }); return { ok: true, status: 200 }; }; + const pending = notify(owner.id, "token_expired", resource); await gate; + await reset(owner.id, resource); await recover(owner.id, resource, async () => {}); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(1); + release(); await pending; + }); + it("rejects stale recovery after another owner action changes the generation", async () => { + await notify(owner.id, "token_expired", resource); await reset(owner.id, resource); + await recover(owner.id, resource, async () => { await reset(owner.id, resource); }); + expect(await guards().findOne({ _id: recipient() })).not.to.equal(null); + }); + it("cancels an unsent claim before a paused sender can start dispatch", async () => { + const collection = mongoose.connection.collection("owner_access_alert_events"), insert = collection.insertOne; + let entered, release; const gate = new Promise(resolve => { entered = resolve; }); + collection.insertOne = async function (...args) { entered(); await new Promise(resolve => { release = resolve; }); return insert.apply(this, args); }; + try { + const pending = notify(owner.id, "token_expired", resource); await gate; + await reset(owner.id, resource); await recover(owner.id, resource, async () => {}); + release(); await pending; expect(sent).to.have.length(0); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + } finally { collection.insertOne = insert; } + }); + it("fences a second owner action that races after recovery's final resource check", async () => { + await notify(owner.id, "token_expired", resource); await reset(owner.id, resource); + const collection = guards(), remove = collection.deleteOne; + let entered, release; const gate = new Promise(resolve => { entered = resolve; }); + collection.deleteOne = async function (...args) { entered(); await new Promise(resolve => { release = resolve; }); return remove.apply(this, args); }; + try { + const pending = recover(owner.id, resource, async () => {}); await gate; + await reset(owner.id, resource); release(); await pending; + expect(await guards().findOne({ _id: recipient() })).not.to.equal(null); + } finally { collection.deleteOne = remove; } + }); + it("does not accept a successful read of a stale source binding", async () => { + await notify(owner.id, "token_expired", resource); await reset(owner.id, resource); + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { "source.repositoryName": "new/source" } }); + await recover(owner.id, resource, async () => {}, { "source.repositoryName": "old/source" }); + expect(await guards().findOne({ _id: recipient() })).not.to.equal(null); + }); + for (const failure of ["rejection", "timeout"]) it(`retains the recipient guard after ${failure}`, async () => { + global.fetch = async (_url, options) => { sent.push(options); if (failure === "timeout") throw new Error("timeout"); return { ok: false, status: 429 }; }; + await notify(owner.id, "token_expired", resource); await reset(owner.id); + await notify(otherOwner.id, "repo_not_found", other); expect(sent).to.have.length(1); + expect((await guards().findOne({ _id: recipient() })).state).to.equal("failed"); + }); + it("releases a guard for a removed resource without erasing delivery history", async () => { + await notify(owner.id, "token_expired", resource); + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { status: "removed" } }); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(2); + expect(await mongoose.connection.collection("owner_access_alert_events").countDocuments()).to.equal(4); + }); + for (const step of ["resource update", "resource verification", "audit insertion", "dispatch transition"]) it(`releases an unsent guard after ${step} fails`, async () => { + const collection = step.startsWith("resource") ? coll(resource.kind) : step === "audit insertion" ? mongoose.connection.collection("owner_access_alert_events") : guards(); + const method = step === "resource verification" ? "findOne" : step === "audit insertion" ? "insertOne" : "updateOne"; + const original = collection[method]; let calls = 0; + collection[method] = async function (...args) { + calls++; + if ((step === "resource verification" && calls === 3) + || (step === "dispatch transition" && args[1].$set?.state === "sending") + || (step === "resource update" && args[1].$set?.accessAlertClaimedAt) + || step === "audit insertion") throw new Error("transient database error"); + return original.apply(this, args); + }; + try { await notify(owner.id, "token_expired", resource); } + finally { collection[method] = original; } + expect(sent).to.have.length(0); expect(await guards().findOne({ _id: recipient() })).to.equal(null); + expect((await coll(resource.kind).findOne({ _id: new mongoose.Types.ObjectId(resource.id) })).accessAlertClaimedAt).to.equal(undefined); + await notify(owner.id, "token_expired", resource); expect(sent).to.have.length(1); + }); + it("retains audit evidence while provider delivery is still in flight", async () => { + let release, entered; const gate = new Promise(resolve => { entered = resolve; }); + global.fetch = async () => { entered(); await new Promise(resolve => { release = resolve; }); return { ok: true, status: 200 }; }; + const pending = notify(owner.id, "token_expired", resource); await gate; + try { + const record = await guards().findOne({ _id: recipient() }); + const event = await mongoose.connection.collection("owner_access_alert_events").findOne({ _id: record.attemptId }); + expect(event.state).to.equal("prepared"); expect(record.state).to.equal("sending"); + } finally { release(); await pending; } + }); + it("reclaims a crashed pre-dispatch attempt without owner action under concurrent failures", async () => { + await guards().insertOne({ _id: recipient(), owner: owner._id, resourceKind: resource.kind, + resourceId: new mongoose.Types.ObjectId(resource.id), attemptId: "crashed-pending", state: "pending", claimedAt: new Date(0) }); + await Promise.all(Array.from({ length: 40 }, (_, i) => notify(i % 2 ? owner.id : otherOwner.id, "token_expired", i % 2 ? resource : other))); + expect(sent).to.have.length(1); + expect((await guards().findOne({ _id: recipient() })).state).to.equal("accepted"); + }); + it("refreshes the resource fence when an old claim is reclaimed before guard insertion", async () => { + await guards().insertOne({ _id: recipient(), owner: owner._id, resourceKind: resource.kind, + resourceId: new mongoose.Types.ObjectId(resource.id), attemptId: "old-pending", state: "pending", claimedAt: new Date(0) }); + const users = mongoose.connection.collection("users"), update = users.updateOne; + let entered, release; const gate = new Promise(resolve => { entered = resolve; }); + users.updateOne = async function (filter, operation, ...args) { + if (Array.isArray(operation)) { entered(); await new Promise(resolve => { release = resolve; }); } + return update.call(this, filter, operation, ...args); + }; + const pending = notify(owner.id, "token_expired", resource); + try { await gate; await alerts.initializeOwnerAccessAlerts(); } + finally { users.updateOne = update; release(); await pending; } + expect(sent).to.have.length(1); + expect((await guards().findOne({ _id: recipient() })).resourceRevision).to.equal(1); + expect((await coll(resource.kind).findOne({ _id: new mongoose.Types.ObjectId(resource.id) })).accessAlertClaimRevision).to.equal(2); + }); + it("keeps recent pending claims and old sending claims blocked", async () => { + await guards().insertOne({ _id: recipient(), owner: owner._id, resourceKind: resource.kind, + resourceId: new mongoose.Types.ObjectId(resource.id), attemptId: "active-pending", state: "pending", claimedAt: new Date() }); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(0); + await guards().updateOne({ _id: recipient() }, { $set: { state: "sending", claimedAt: new Date(0) } }); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(0); + await alerts.initializeOwnerAccessAlerts(); + expect((await guards().findOne({ _id: recipient() })).state).to.equal("sending"); + }); + it("reaps only expired unsent claims during startup and never reseeds them", async () => { + await guards().insertOne({ _id: "recipient-alerts-v1", state: "ready" }); + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { accessAlertClaimedAt: new Date(0), accessAlertAttemptId: recipient() } }); + for (const [key, state, date] of [[recipient(), "pending", new Date(0)], ["recent", "pending", new Date()], + ["sending", "sending", new Date(0)], ["accepted", "accepted", new Date(0)], ["failed", "failed", new Date(0)]]) { + await guards().insertOne({ _id: key, owner: owner._id, resourceKind: resource.kind, + resourceId: new mongoose.Types.ObjectId(resource.id), attemptId: key, state, claimedAt: date }); + } + await alerts.initializeOwnerAccessAlerts(); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + for (const key of ["recent", "sending", "accepted", "failed"]) expect(await guards().findOne({ _id: key })).not.to.equal(null); + await alerts.initializeOwnerAccessAlerts(); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + await notify(owner.id, "token_expired", resource); expect(sent).to.have.length(1); + }); + it("fences a paused sender after another reader reclaims its expired pending claim", async () => { + const collection = mongoose.connection.collection("owner_access_alert_events"), insert = collection.insertOne; + let entered, release; const gate = new Promise(resolve => { entered = resolve; }); + let pausedOnce = false; + collection.insertOne = async function (...args) { + if (!pausedOnce) { pausedOnce = true; entered(); await new Promise(resolve => { release = resolve; }); } + return insert.apply(this, args); + }; + const paused = notify(owner.id, "token_expired", resource); + try { + await gate; + await guards().updateOne({ _id: recipient() }, { $set: { claimedAt: new Date(0) } }); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(1); + } finally { collection.insertOne = insert; release(); await paused; } + expect(sent).to.have.length(1); + expect((await guards().findOne({ _id: recipient() })).owner.toString()).to.equal(otherOwner.id); + }); + it("cannot reclaim a pending attempt that starts dispatch before cancellation", async () => { + await guards().insertOne({ _id: recipient(), owner: owner._id, resourceKind: resource.kind, + resourceId: new mongoose.Types.ObjectId(resource.id), attemptId: "racing-dispatch", state: "pending", claimedAt: new Date(0) }); + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { accessAlertClaimedAt: new Date(0), accessAlertAttemptId: "racing-dispatch" } }); + const collection = guards(), update = collection.updateOne; + collection.updateOne = async function (filter, operation, ...args) { + if (operation.$set?.state === "cancelled") await update.call(this, { _id: recipient(), state: "pending", attemptId: "racing-dispatch" }, { $set: { state: "sending" } }); + return update.call(this, filter, operation, ...args); + }; + try { await notify(otherOwner.id, "token_expired", other); } + finally { collection.updateOne = update; } + expect(sent).to.have.length(0); expect((await guards().findOne({ _id: recipient() })).state).to.equal("sending"); + expect((await coll(resource.kind).findOne({ _id: new mongoose.Types.ObjectId(resource.id) })).accessAlertAttemptId).to.equal("racing-dispatch"); + }); + for (const state of ["pending", "cancelled"]) it(`clears a crashed unsent ${state} attempt's legacy fields before migration`, async () => { + const attemptId = `crashed-${state}`; + await guards().insertOne({ _id: recipient(), owner: owner._id, resourceKind: resource.kind, + resourceId: new mongoose.Types.ObjectId(resource.id), generation: "old-generation", attemptId, state, + claimedAt: state === "pending" ? new Date(0) : new Date() }); + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { + accessAlertGeneration: "old-generation", accessAlertClaimedAt: new Date(0), accessAlertAttemptId: attemptId, + } }); + await alerts.initializeOwnerAccessAlerts(); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + const row = await coll(resource.kind).findOne({ _id: new mongoose.Types.ObjectId(resource.id) }); + expect(row.accessAlertClaimedAt).to.equal(undefined); expect(row.accessAlertAttemptId).to.equal(undefined); + expect(sent).to.have.length(0); + await notify(owner.id, "token_expired", resource); expect(sent).to.have.length(1); + }); + for (const changed of ["generation", "attempt"]) it(`preserves newer legacy fields when the ${changed} no longer matches the cancelled attempt`, async () => { + await guards().insertOne({ _id: "recipient-alerts-v1", state: "ready" }); + await guards().insertOne({ _id: recipient(), owner: owner._id, resourceKind: resource.kind, + resourceId: new mongoose.Types.ObjectId(resource.id), generation: "old-generation", attemptId: "old-attempt", state: "pending", claimedAt: new Date(0) }); + const generation = changed === "generation" ? "new-generation" : "old-generation"; + const attemptId = changed === "attempt" ? "new-attempt" : "old-attempt"; + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { + accessAlertGeneration: generation, accessAlertClaimedAt: new Date(1234), accessAlertAttemptId: attemptId, + } }); + await alerts.initializeOwnerAccessAlerts(); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + const row = await coll(resource.kind).findOne({ _id: new mongoose.Types.ObjectId(resource.id) }); + expect(row.accessAlertClaimedAt.getTime()).to.equal(1234); expect(row.accessAlertAttemptId).to.equal(attemptId); + }); + it("resumes cancelled metadata cleanup after a database failure without creating false legacy evidence", async () => { + const collection = coll(resource.kind), update = collection.updateOne, read = collection.findOne; + let reads = 0; + collection.findOne = async function (...args) { if (++reads === 3) throw new Error("verification failed"); return read.apply(this, args); }; + collection.updateOne = async function (filter, operation, ...args) { + if (Array.isArray(operation)) throw new Error("cleanup unavailable"); + return update.call(this, filter, operation, ...args); + }; + try { await notify(owner.id, "token_expired", resource); } + finally { collection.findOne = read; collection.updateOne = update; } + expect(sent).to.have.length(0); expect((await guards().findOne({ _id: recipient() })).state).to.equal("cancelled"); + await alerts.initializeOwnerAccessAlerts(); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + expect((await collection.findOne({ _id: new mongoose.Types.ObjectId(resource.id) })).accessAlertClaimedAt).to.equal(undefined); + await notify(owner.id, "token_expired", resource); expect(sent).to.have.length(1); + }); + it("cleans recent pending legacy fields once deployment has stopped every writer", async () => { + await guards().insertOne({ _id: "recipient-alerts-v1", state: "ready" }); + await guards().insertOne({ _id: recipient(), owner: owner._id, resourceKind: resource.kind, + resourceId: new mongoose.Types.ObjectId(resource.id), attemptId: "recent-unsent", state: "pending", claimedAt: new Date() }); + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { + accessAlertClaimedAt: new Date(), accessAlertAttemptId: "recent-unsent", + } }); + await alerts.prepareLegacyOwnerAlertMigration(); await alerts.initializeOwnerAccessAlerts(); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + expect((await coll(resource.kind).findOne({ _id: new mongoose.Types.ObjectId(resource.id) })).accessAlertClaimedAt).to.equal(undefined); + }); + it("preserves an untagged legacy delivery written during rollback while cancelling a native pending attempt", async () => { + await guards().insertOne({ _id: "recipient-alerts-v1", state: "ready" }); + await guards().insertOne({ _id: recipient(), owner: owner._id, resourceKind: resource.kind, + resourceId: new mongoose.Types.ObjectId(resource.id), attemptId: "native-unsent", state: "pending", claimedAt: new Date() }); + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { accessAlertClaimedAt: new Date() } }); + await alerts.prepareLegacyOwnerAlertMigration(); await alerts.initializeOwnerAccessAlerts(); + expect((await guards().findOne({ _id: recipient() })).state).to.equal("legacy"); + await notify(owner.id, "token_expired", resource); expect(sent).to.have.length(0); + await reset(owner.id, resource); await recover(owner.id, resource, async () => {}); + await notify(owner.id, "token_expired", resource); expect(sent).to.have.length(1); + }); + it("blocks dispatch while an owner reset has advanced the guard revision but not the resource generation", async () => { + const events = mongoose.connection.collection("owner_access_alert_events"), insert = events.insertOne; + const collection = coll(resource.kind), updateMany = collection.updateMany; + let prepared, releasePrepared, resetting, releaseReset; + const atPrepared = new Promise(resolve => { prepared = resolve; }); + const atReset = new Promise(resolve => { resetting = resolve; }); + events.insertOne = async function (...args) { prepared(); await new Promise(resolve => { releasePrepared = resolve; }); return insert.apply(this, args); }; + collection.updateMany = async function (...args) { resetting(); await new Promise(resolve => { releaseReset = resolve; }); return updateMany.apply(this, args); }; + const pending = notify(owner.id, "token_expired", resource); + let action; + try { + await atPrepared; action = reset(owner.id, resource); await atReset; + expect((await guards().findOne({ _id: recipient() })).resetRevision).to.equal(1); + releasePrepared(); await pending; + expect(sent).to.have.length(0); + const row = await collection.findOne({ _id: new mongoose.Types.ObjectId(resource.id) }); + expect(row.accessAlertGeneration).to.equal(undefined); expect(row.accessAlertClaimedAt).to.equal(undefined); + } finally { + events.insertOne = insert; collection.updateMany = updateMany; + if (releasePrepared) releasePrepared(); if (releaseReset) releaseReset(); + await pending; if (action) await action; + } + await notify(owner.id, "token_expired", resource); expect(sent).to.have.length(1); + }); + it("invalidates an old failure when reset starts before its recipient guard exists", async () => { + const lookup = Users.findOne, collection = coll(resource.kind), updateMany = collection.updateMany; + let entered, release, resetting, releaseReset; + const atLookup = new Promise(resolve => { entered = resolve; }); + const atReset = new Promise(resolve => { resetting = resolve; }); + Users.findOne = function (...args) { return { lean: async () => { + entered(); await new Promise(resolve => { release = resolve; }); return lookup.apply(Users, args).lean(); + } }; }; + collection.updateMany = async function (...args) { resetting(); await new Promise(resolve => { releaseReset = resolve; }); return updateMany.apply(this, args); }; + const pending = notify(owner.id, "token_expired", resource); let action; + try { + await atLookup; action = reset(owner.id, resource); await atReset; + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + release(); await pending; expect(sent).to.have.length(0); + expect((await collection.findOne({ _id: new mongoose.Types.ObjectId(resource.id) })).accessAlertClaimedAt).to.equal(undefined); + } finally { + Users.findOne = lookup; collection.updateMany = updateMany; + if (release) release(); if (releaseReset) releaseReset(); await pending; if (action) await action; + } + await notify(owner.id, "token_expired", resource); expect(sent).to.have.length(1); + }); + it("prevents a cancelled sender from recreating legacy fields after its database write resumes", async () => { + const collection = coll(resource.kind), update = collection.updateOne; + let entered, release; const gate = new Promise(resolve => { entered = resolve; }); + collection.updateOne = async function (filter, operation, ...args) { + if (operation.$set?.accessAlertClaimedAt) { entered(); await new Promise(resolve => { release = resolve; }); } + return update.call(this, filter, operation, ...args); + }; + const pending = notify(owner.id, "token_expired", resource); + try { + await gate; await guards().updateOne({ _id: recipient() }, { $set: { claimedAt: new Date(0) } }); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(1); + } finally { collection.updateOne = update; release(); await pending; } + const row = await collection.findOne({ _id: new mongoose.Types.ObjectId(resource.id) }); + expect(row.accessAlertClaimedAt).to.equal(undefined); expect(row.accessAlertClaimRevision).to.equal(1); + expect(sent).to.have.length(1); + }); + it("reaps a crashed pre-provider attempt after legacy preparation completed", async () => { + await guards().insertOne({ _id: recipient(), owner: owner._id, resourceKind: resource.kind, + resourceId: new mongoose.Types.ObjectId(resource.id), attemptId: "prepared-not-sent", state: "pending", resourceRevision: 0, claimedAt: new Date(0) }); + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { + accessAlertClaimedAt: new Date(0), accessAlertAttemptId: "prepared-not-sent", accessAlertClaimRevision: 1, + } }); + await alerts.initializeOwnerAccessAlerts(); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); expect(sent).to.have.length(0); + const row = await coll(resource.kind).findOne({ _id: new mongoose.Types.ObjectId(resource.id) }); + expect(row.accessAlertClaimedAt).to.equal(undefined); expect(row.accessAlertClaimRevision).to.equal(2); + await notify(owner.id, "token_expired", resource); expect(sent).to.have.length(1); + }); + for (const accepted of [true, false]) it(`applies recovery recorded during an in-flight ${accepted ? "accepted" : "rejected"} delivery`, async () => { + let entered, release; const gate = new Promise(resolve => { entered = resolve; }); + global.fetch = async (_url, options) => { + sent.push(options); if (sent.length === 1) { entered(); await new Promise(resolve => { release = resolve; }); } + return { ok: accepted, status: accepted ? 200 : 403 }; + }; + const pending = notify(owner.id, "token_expired", resource); + try { + await gate; await reset(owner.id, resource); await recover(owner.id, resource, async () => {}); + const record = await guards().findOne({ _id: recipient() }); expect(record.state).to.equal("sending"); + expect(record.deferredRecoveries).to.have.length(1); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(1); + } finally { release(); await pending; } + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(2); + }); + it("invalidates in-flight recovery evidence after a newer reset", async () => { + let entered, release; const gate = new Promise(resolve => { entered = resolve; }); + global.fetch = async (_url, options) => { + sent.push(options); if (sent.length === 1) { entered(); await new Promise(resolve => { release = resolve; }); } + return { ok: true, status: 200 }; + }; + const pending = notify(owner.id, "token_expired", resource); + try { + await gate; await reset(owner.id, resource); await recover(owner.id, resource, async () => {}); await reset(owner.id, resource); + } finally { release(); await pending; } + expect((await guards().findOne({ _id: recipient() })).state).to.equal("accepted"); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(1); + await recover(owner.id, resource, async () => {}); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(2); + }); + it("invalidates deferred recovery if its source binding changes before delivery completes", async () => { + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { source: { repositoryName: "owner/repo" } } }); + let entered, release; const gate = new Promise(resolve => { entered = resolve; }); + global.fetch = async (_url, options) => { + sent.push(options); if (sent.length === 1) { entered(); await new Promise(resolve => { release = resolve; }); } + return { ok: true, status: 200 }; + }; + const pending = notify(owner.id, "token_expired", resource); + try { + await gate; await reset(owner.id, resource); + await recover(owner.id, resource, async () => {}, { "source.repositoryName": "owner/repo" }); + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { "source.repositoryName": "new/repo" } }); + } finally { release(); await pending; } + expect((await guards().findOne({ _id: recipient() })).state).to.equal("accepted"); + await recover(owner.id, resource, async () => {}, { "source.repositoryName": "new/repo" }); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(2); + }); + it("rejects dispatch when reset starts after the guard reservation but before the owner commitment", async () => { + const collection = guards(), update = collection.updateOne, updateMany = collection.updateMany; + let entered, release, resetting, releaseReset; + const gate = new Promise(resolve => { entered = resolve; }), resetGate = new Promise(resolve => { resetting = resolve; }); + collection.updateOne = async function (filter, operation, ...args) { + const result = await update.call(this, filter, operation, ...args); + if (operation.$set?.state === "sending") { entered(); await new Promise(resolve => { release = resolve; }); } + return result; + }; + collection.updateMany = async function (...args) { resetting(); await new Promise(resolve => { releaseReset = resolve; }); return updateMany.apply(this, args); }; + const pending = notify(owner.id, "token_expired", resource); let action; + try { + await gate; action = reset(owner.id, resource); await resetGate; + expect((await guards().findOne({ _id: recipient() })).state).to.equal("sending"); + release(); await pending; expect(sent).to.have.length(0); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + } finally { + collection.updateOne = update; collection.updateMany = updateMany; + if (release) release(); if (releaseReset) releaseReset(); await pending; if (action) await action; + } + await notify(owner.id, "token_expired", resource); expect(sent).to.have.length(1); + }); + it("applies fresh recovery recorded before the reset lease is cleared without another upstream read", async () => { + await notify(owner.id, "token_expired", resource); + const users = mongoose.connection.collection("users"), update = users.updateOne; + let entered, release; const gate = new Promise(resolve => { entered = resolve; }); + users.updateOne = async function (filter, operation, ...args) { + if (operation.$unset?.accessAlertResetUntil !== undefined) { entered(); await new Promise(resolve => { release = resolve; }); } + return update.call(this, filter, operation, ...args); + }; + const action = reset(owner.id, resource); + try { + await gate; await recover(owner.id, resource, async () => {}); + expect((await guards().findOne({ _id: recipient() })).deferredRecoveries).to.have.length(1); + } finally { users.updateOne = update; release(); await action; } + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(2); + }); + it("retries saved recovery after a crashed reset lease expires", async () => { + await notify(owner.id, "token_expired", resource); await reset(owner.id, resource); + await mongoose.connection.collection("users").updateOne({ _id: owner._id }, { $set: { accessAlertResetUntil: new Date(Date.now() + 150) } }); + await recover(owner.id, resource, async () => {}); + expect(await guards().findOne({ _id: recipient() })).not.to.equal(null); + for (let i = 0; i < 100 && await guards().findOne({ _id: recipient() }); i++) await new Promise(resolve => global.setTimeout(resolve, 20)); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(2); + }); + for (const state of ["prepared", "cancelled", "committed"]) it(`reconciles interrupted guard reservation with a ${state} owner dispatch record`, async () => { + const attemptId = `interrupted-${state}`; + await guards().insertMany([{ _id: "recipient-alerts-v1", state: "ready" }, { + _id: recipient(), owner: owner._id, resourceKind: resource.kind, resourceId: new mongoose.Types.ObjectId(resource.id), + attemptId, state: "sending", dispatchPrepared: true, resourceRevision: 0, claimedAt: new Date(0), + }]); + await mongoose.connection.collection("users").updateOne({ _id: owner._id }, { $set: { + [`accessAlertDispatches.${attemptId}`]: { state, claimedAt: new Date(0) }, + } }); + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { + accessAlertClaimedAt: new Date(0), accessAlertAttemptId: attemptId, accessAlertClaimRevision: 1, + } }); + await alerts.initializeOwnerAccessAlerts(); + const record = await guards().findOne({ _id: recipient() }); + const account = await mongoose.connection.collection("users").findOne({ _id: owner._id }); + if (state === "committed") { + expect(record.state).to.equal("sending"); expect(account.accessAlertDispatches[attemptId].state).to.equal(state); + } else { + expect(record).to.equal(null); expect(account.accessAlertDispatches[attemptId]).to.equal(undefined); + expect((await coll(resource.kind).findOne({ _id: new mongoose.Types.ObjectId(resource.id) })).accessAlertClaimedAt).to.equal(undefined); + } + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(state === "committed" ? 0 : 1); + }); + it("prunes expired uncommitted owner dispatch records and cleans completed attempts", async () => { + await mongoose.connection.collection("users").updateOne({ _id: owner._id }, { $set: { accessAlertDispatches: { + abandoned: { state: "prepared", claimedAt: new Date(0) }, cancelled: { state: "cancelled", claimedAt: new Date(0) }, + unknown: { state: "committed", claimedAt: new Date(0) }, + } } }); + await notify(owner.id, "token_expired", resource); expect(sent).to.have.length(1); + const account = await mongoose.connection.collection("users").findOne({ _id: owner._id }); + expect(Object.keys(account.accessAlertDispatches)).to.deep.equal(["unknown"]); + }); + it("allows active accounts to receive alerts when a shared-address guard belongs to a banned owner", async () => { + await notify(owner.id, "token_expired", resource); + await Users.updateOne({ _id: owner._id }, { $set: { status: "banned" } }); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(2); + }); + it("resumes an indexed migration from its last completed checkpoint", async () => { + await Users.updateOne({ _id: otherOwner._id }, { $set: { notificationEmail: "second@example.com" } }); + const second = await create(otherOwner._id); + await coll("repository").updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { accessAlertClaimedAt: new Date(1) } }); + await coll("repository").updateOne({ _id: new mongoose.Types.ObjectId(second.id) }, { $set: { accessAlertClaimedAt: new Date(2) } }); + await guards().insertOne({ _id: "recipient-alerts-v1", state: "new", leaseUntil: new Date(0), + progress: { repository: { claimedAt: new Date(1), id: new mongoose.Types.ObjectId(resource.id) } } }); + await alerts.initializeOwnerAccessAlerts(); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + expect(await guards().findOne({ _id: alerts.recipientAlertKey("second@example.com") })).not.to.equal(null); + expect((await guards().findOne({ _id: "recipient-alerts-v1" })).progress.repository.claimedAt.getTime()).to.equal(2); + expect((await coll("repository").indexes()).some(i => i.name === "owner_access_alert_backfill" && i.partialFilterExpression)).to.equal(true); + }); + it("confirms recovery on a fresh uncached file read but never on a cache hit", async () => { + const database = require("../src/server/database"); + const GitHubStream = require("../src/core/source/GitHubStream").default; + const storage = require("../src/core/storage").default; + const { Readable } = require("stream"); + const originals = { isConnected: database.isConnected, fileInfo: storage.fileInfo, write: storage.write, read: storage.read }; + await coll("repository").updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { repoId: resource.id, source: { repositoryName: "owner/repo", commit: "pinned" } } }); + let cached = false; + database.isConnected = true; + storage.fileInfo = async () => { if (!cached) throw Object.assign(new Error("miss"), { code: "ENOENT" }); return { size: 11, contentType: "text/plain" }; }; + storage.write = async (_repo, _path, input) => { for await (const _chunk of input) { /* Consume the upstream stream. */ } cached = true; }; + storage.read = async () => Readable.from(["fresh-bytes"]); + try { + await notify(owner.id, "token_expired", resource); await reset(owner.id, resource); + const source = new GitHubStream({ repoId: resource.id, organization: "owner", repoName: "repo", commit: "pinned", getToken: async () => "mock-token" }); + source.downloadWithFallback = async () => Readable.from(["fresh-bytes"]); + const read = () => source.getFileContentCache("a.txt", resource.id, () => ({ sha: "a".repeat(40), size: 11 })); + (await read()).destroy(); expect(await guards().findOne({ _id: recipient() })).to.equal(null); + await notify(owner.id, "token_expired", resource); await reset(owner.id, resource); + (await read()).destroy(); expect(await guards().findOne({ _id: recipient() })).not.to.equal(null); + } finally { database.isConnected = originals.isConnected; storage.fileInfo = originals.fileInfo; storage.write = originals.write; storage.read = originals.read; } + }); + it("confirms recovery when a successful scheduled refresh finds the same commit", async () => { + const Repository = require("../src/core/Repository").default; + const Model = require("../src/core/model/anonymizedRepositories/anonymizedRepositories.model").default; + const github = require("../src/core/source/GitHubRepository"); + const original = github.getRepositoryFromGitHub; + const source = { repositoryName: "owner/repo", commit: "pinned", branch: "main" }; + await coll("repository").updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { source } }); + const repo = new Repository(new Model({ _id: resource.id, owner: owner._id, repoId: resource.id, status: "ready", + source, lastView: new Date(0), options: { update: true, expirationMode: "never" } })); + repo.getToken = async () => "mock-token"; + github.getRepositoryFromGitHub = async () => ({ fullName: "owner/repo", model: { defaultBranch: "main" }, + branches: async () => [{ name: "main", commit: "pinned" }], getCommitInfo: async () => { throw new Error("Unexpected commit fetch"); } }); + try { + await notify(owner.id, "token_expired", resource); await reset(owner.id, resource); + await repo.updateIfNeeded({ force: false }); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + } finally { github.getRepositoryFromGitHub = original; } + }); + it("seeds old claims once while five instances start concurrently", async () => { + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { accessAlertClaimedAt: new Date() } }); + await Promise.all(Array.from({ length: 5 }, () => alerts.initializeOwnerAccessAlerts())); + expect((await guards().findOne({ _id: recipient() })).state).to.equal("legacy"); + await notify(otherOwner.id, "repo_not_found", other); expect(sent).to.have.length(0); + await reset(owner.id, resource); await recover(owner.id, resource, async () => {}); + await alerts.initializeOwnerAccessAlerts(); expect((await guards().findOne({ _id: recipient() })).state).to.equal("recovered"); + }); + it("preserves every legacy claim for a shared address until each triggering resource recovers", async () => { + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { accessAlertClaimedAt: new Date(1) } }); + await coll(other.kind).updateOne({ _id: new mongoose.Types.ObjectId(other.id) }, { $set: { accessAlertClaimedAt: new Date(2) } }); + await alerts.initializeOwnerAccessAlerts(); + expect((await guards().findOne({ _id: recipient() })).legacyClaims).to.have.length(2); + await reset(owner.id, resource); await recover(owner.id, resource, async () => {}); + expect((await guards().findOne({ _id: recipient() })).legacyClaims.filter(claim => !claim.recovered)).to.have.length(1); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(0); + await reset(otherOwner.id, other); await recover(otherOwner.id, other, async () => {}); + expect((await guards().findOne({ _id: recipient() })).state).to.equal("recovered"); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(1); + }); + it("does not retire a legacy group when only its first triggering resource is removed", async () => { + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { accessAlertClaimedAt: new Date(1) } }); + await coll(other.kind).updateOne({ _id: new mongoose.Types.ObjectId(other.id) }, { $set: { accessAlertClaimedAt: new Date(2) } }); + await alerts.initializeOwnerAccessAlerts(); + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { status: "removed" } }); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(0); + await reset(otherOwner.id, other); await recover(otherOwner.id, other, async () => {}); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(1); + }); + it("atomically releases the final legacy guard during concurrent recoveries", async () => { + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { accessAlertClaimedAt: new Date(1) } }); + await coll(other.kind).updateOne({ _id: new mongoose.Types.ObjectId(other.id) }, { $set: { accessAlertClaimedAt: new Date(2) } }); + await alerts.initializeOwnerAccessAlerts(); await reset(owner.id, resource); await reset(otherOwner.id, other); + await Promise.all([recover(owner.id, resource, async () => {}), recover(otherOwner.id, other, async () => {})]); + expect((await guards().findOne({ _id: recipient() })).state).to.equal("recovered"); + }); + it("requires a fresh read after a newer reset of a partially recovered legacy group", async () => { + for (const item of [resource, other]) await coll(item.kind).updateOne({ _id: new mongoose.Types.ObjectId(item.id) }, { $set: { accessAlertClaimedAt: new Date(1) } }); + await alerts.initializeOwnerAccessAlerts(); + await reset(owner.id, resource); await recover(owner.id, resource, async () => {}); + await reset(owner.id, resource); + await reset(otherOwner.id, other); await recover(otherOwner.id, other, async () => {}); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(0); + expect((await guards().findOne({ _id: recipient() })).state).to.equal("legacy"); + await recover(owner.id, resource, async () => {}); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(1); + }); + it("fences legacy claim mutation when reset races after the final resource check", async () => { + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { accessAlertClaimedAt: new Date(1) } }); + await alerts.initializeOwnerAccessAlerts(); await reset(owner.id, resource); + const collection = guards(), update = collection.updateOne; + let entered, release; const gate = new Promise(resolve => { entered = resolve; }); + collection.updateOne = async function (filter, operation, ...args) { + if (filter.attemptId && Array.isArray(operation)) { entered(); await new Promise(resolve => { release = resolve; }); } + return update.call(this, filter, operation, ...args); + }; + const pending = recover(owner.id, resource, async () => {}); + try { await gate; await reset(owner.id, resource); } + finally { collection.updateOne = update; release(); await pending; } + const record = await guards().findOne({ _id: recipient() }); + expect(record.state).to.equal("legacy"); expect(record.legacyClaims[0].recovered).to.equal(false); + }); + it("reconciles claims written by legacy instances after rollback without reseeding ordinary restarts", async () => { + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { accessAlertClaimedAt: new Date(1) } }); + await alerts.initializeOwnerAccessAlerts(); await reset(owner.id, resource); await recover(owner.id, resource, async () => {}); + await alerts.initializeOwnerAccessAlerts(); expect((await guards().findOne({ _id: recipient() })).state).to.equal("recovered"); + await coll(other.kind).updateOne({ _id: new mongoose.Types.ObjectId(other.id) }, { $set: { accessAlertClaimedAt: new Date() } }); + await alerts.prepareLegacyOwnerAlertMigration(); + const marker = await guards().findOne({ _id: "recipient-alerts-v1" }); expect(marker.state).to.equal("new"); expect(marker.progress).to.equal(undefined); + await alerts.initializeOwnerAccessAlerts(); await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(0); + expect((await guards().findOne({ _id: recipient() })).legacyClaims.filter(claim => !claim.recovered)).to.have.length(1); + await reset(otherOwner.id, other); await recover(otherOwner.id, other, async () => {}); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(1); + }); + it("preserves a native guard's origin when rollback reconciliation adds another legacy claim", async () => { + await alerts.initializeOwnerAccessAlerts(); await notify(owner.id, "token_expired", resource); await reset(owner.id, resource); + await coll(other.kind).updateOne({ _id: new mongoose.Types.ObjectId(other.id) }, { $set: { accessAlertClaimedAt: new Date() } }); + await alerts.prepareLegacyOwnerAlertMigration(); await alerts.initializeOwnerAccessAlerts(); + expect((await guards().findOne({ _id: recipient() })).legacyClaims.filter(claim => !claim.recovered)).to.have.length(2); + await reset(otherOwner.id, other); await recover(otherOwner.id, other, async () => {}); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(1); + await recover(owner.id, resource, async () => {}); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(2); + }); + it("fences completed-guard retirement against a concurrent reactivation and reset", async () => { + await notify(owner.id, "token_expired", resource); + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { status: "removed" } }); + const collection = guards(), remove = collection.deleteOne; + let entered, release; const gate = new Promise(resolve => { entered = resolve; }); + collection.deleteOne = async function (...args) { entered(); await new Promise(resolve => { release = resolve; }); return remove.apply(this, args); }; + const pending = notify(otherOwner.id, "token_expired", other); + try { + await gate; + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { status: "ready" } }); + await reset(owner.id, resource); + } finally { collection.deleteOne = remove; release(); await pending; } + expect(sent).to.have.length(1); expect(await guards().findOne({ _id: recipient() })).not.to.equal(null); + await recover(owner.id, resource, async () => {}); await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(2); + }); + for (const accepted of [true, false]) it(`retries the guard update after a definite ${accepted ? "accepted" : "rejected"} provider response`, async () => { + const collection = guards(), update = collection.updateOne; let writes = 0; + global.fetch = async (_url, options) => { sent.push(options); return { ok: accepted, status: accepted ? 200 : 403 }; }; + collection.updateOne = async function (filter, operation, ...args) { + if (["accepted", "failed"].includes(operation.$set?.state)) { + expect(await mongoose.connection.collection("owner_access_alert_events").findOne({ _id: `${filter.attemptId}/completion` })).not.to.equal(null); + if (++writes === 1) throw new Error("transient result write failure"); + } + return update.call(this, filter, operation, ...args); + }; + try { await notify(owner.id, "token_expired", resource); } + finally { collection.updateOne = update; } + expect(writes).to.equal(2); expect(sent).to.have.length(1); + expect((await guards().findOne({ _id: recipient() })).state).to.equal(accepted ? "accepted" : "failed"); + }); + for (const startup of [true, false]) it(`reconciles a durable provider result during ${startup ? "startup" : "the next notification"}`, async () => { + await alerts.initializeOwnerAccessAlerts(); + const collection = guards(), update = collection.updateOne; let writes = 0; + collection.updateOne = async function (filter, operation, ...args) { + if (["accepted", "failed"].includes(operation.$set?.state)) { writes++; throw new Error("database unavailable"); } + return update.call(this, filter, operation, ...args); + }; + try { await notify(owner.id, "token_expired", resource); } + finally { collection.updateOne = update; } + expect(writes).to.equal(3); expect((await guards().findOne({ _id: recipient() })).state).to.equal("sending"); + if (startup) await alerts.initializeOwnerAccessAlerts(); else await notify(otherOwner.id, "token_expired", other); + expect((await guards().findOne({ _id: recipient() })).state).to.equal("accepted"); expect(sent).to.have.length(1); + await reset(owner.id, resource); await recover(owner.id, resource, async () => {}); + await notify(otherOwner.id, "token_expired", other); expect(sent).to.have.length(2); + }); + it("records a provider completion once after its insertion acknowledgement is lost", async () => { + const collection = mongoose.connection.collection("owner_access_alert_events"), insert = collection.insertOne; + let lost = false; + collection.insertOne = async function (record, ...args) { + const result = await insert.call(this, record, ...args); + if (record.state === "accepted" && !lost) { lost = true; throw new Error("lost acknowledgement"); } + return result; + }; + try { await notify(owner.id, "token_expired", resource); } + finally { collection.insertOne = insert; } + expect(lost).to.equal(true); expect(sent).to.have.length(1); + expect(await collection.countDocuments({ state: "accepted" })).to.equal(1); + expect((await guards().findOne({ _id: recipient() })).state).to.equal("accepted"); + }); + it("does not let a late provider-result write resurrect a recovered guard", async () => { + const collection = guards(), update = collection.updateOne; + let entered, release, paused = false; const gate = new Promise(resolve => { entered = resolve; }); + collection.updateOne = async function (filter, operation, ...args) { + if (operation.$set?.state === "accepted" && !paused) { paused = true; entered(); await new Promise(resolve => { release = resolve; }); } + return update.call(this, filter, operation, ...args); + }; + const pending = notify(owner.id, "token_expired", resource); + try { + await gate; await notify(otherOwner.id, "token_expired", other); + await reset(owner.id, resource); await recover(owner.id, resource, async () => {}); + } finally { collection.updateOne = update; release(); await pending; } + expect(sent).to.have.length(1); expect(await guards().findOne({ _id: recipient() })).to.equal(null); + }); + it("confirms truncated-file metadata recovery only on a successful fresh matching-source probe", async () => { + const database = require("../src/server/database"); + const GitHubStream = require("../src/core/source/GitHubStream").default; + const utils = require("../src/core/GitHubUtils"); + const originals = { isConnected: database.isConnected, octokit: utils.octokit }; + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { repoId: resource.id, source: { repositoryName: "owner/repo", commit: "pinned" } } }); + let response = { type: "file", sha: "a".repeat(40), size: 11 }; + database.isConnected = true; + utils.octokit = () => ({ repos: { getContent: async () => { if (response instanceof Error) throw response; return { data: response }; } } }); + const source = new GitHubStream({ repoId: resource.id, organization: "owner", repoName: "repo", commit: "pinned", getToken: async () => "mock-token" }); + try { + await notify(owner.id, "token_expired", resource); await reset(owner.id, resource); + expect((await source.fetchFileInfoFromPath("a.txt")).sha).to.equal("a".repeat(40)); + expect(await guards().findOne({ _id: recipient() })).to.equal(null); + await notify(owner.id, "token_expired", resource); await reset(owner.id, resource); + response = { type: "dir" }; expect(await source.fetchFileInfoFromPath("a.txt")).to.equal(null); + expect(await guards().findOne({ _id: recipient() })).not.to.equal(null); + response = Object.assign(new Error("Forbidden"), { status: 403 }); + try { await source.fetchFileInfoFromPath("a.txt"); throw new Error("Expected a failed probe"); } + catch (error) { expect(error.message).to.equal("Forbidden"); } + expect(await guards().findOne({ _id: recipient() })).not.to.equal(null); + response = { type: "file", sha: "a".repeat(40), size: 11 }; + await coll(resource.kind).updateOne({ _id: new mongoose.Types.ObjectId(resource.id) }, { $set: { "source.repositoryName": "different/source" } }); + await source.fetchFileInfoFromPath("a.txt"); + expect(await guards().findOne({ _id: recipient() })).not.to.equal(null); + } finally { database.isConnected = originals.isConnected; utils.octokit = originals.octokit; } + }); + it("resumes a migration after a startup crashed and its lease expired", async () => { + await guards().insertOne({ _id: "recipient-alerts-v1", state: "new", lease: "crashed", leaseUntil: new Date(0) }); + await alerts.initializeOwnerAccessAlerts(); expect((await guards().findOne({ _id: "recipient-alerts-v1" })).state).to.equal("ready"); + }); + it("continues waiting beyond 300 polls while another migration holds a healthy lease", async () => { + await guards().insertOne({ _id: "recipient-alerts-v1", state: "new", lease: "active-migration", leaseUntil: new Date(Date.now() + 600000) }); + const collection = guards(), update = collection.updateOne, timeout = global.setTimeout; + let polls = 0; + collection.updateOne = async function (filter, ...args) { + if (filter._id === "recipient-alerts-v1" && filter.$or && ++polls === 305) { + await update.call(this, { _id: "recipient-alerts-v1" }, { $set: { state: "ready" } }); + } + return update.call(this, filter, ...args); + }; + global.setTimeout = (callback, delay, ...args) => delay === 250 ? timeout(callback, 0, ...args) : timeout(callback, delay, ...args); + try { await alerts.initializeOwnerAccessAlerts(); } + finally { collection.updateOne = update; global.setTimeout = timeout; } + expect(polls).to.equal(305); + expect((await guards().findOne({ _id: "recipient-alerts-v1" })).state).to.equal("ready"); + }); +}); diff --git a/test/registration-email.test.js b/test/registration-email.test.js index 4cc6432..88b5176 100644 --- a/test/registration-email.test.js +++ b/test/registration-email.test.js @@ -17,9 +17,9 @@ const verified = (email, primary = true) => ({ email, primary, verified: true }) describe("registration email collection", function () { const restores = []; const stub = (object, key, value) => { const old = object[key]; restores.push(() => object[key] = old); object[key] = value; }; - let response, calls, writes, user, saved, login; + let response, calls, writes, user, saved, login, resetCalls; beforeEach(() => { - response = [verified("primary@example.com")]; calls = []; writes = []; saved = []; + response = [verified("primary@example.com")]; calls = []; writes = []; saved = []; resetCalls = []; user = new Users({ username: "owner", externalIDs: { github: "10" }, emails: [] }); stub(global, "fetch", async (url, options) => { calls.push({ url, options }); @@ -29,7 +29,7 @@ describe("registration email collection", function () { stub(Users, "findOne", async () => user); stub(Users.prototype, "save", async function () { saved.push(this); return this; }); stub(credentials, "setCredential", async () => {}); - stub(notifications, "resetOwnerAccessAlerts", async () => {}); + stub(notifications, "resetOwnerAccessAlerts", async id => { resetCalls.push(id); }); stub(github, "saveAppGrant", async () => {}); stub(github, "exchangeAppToken", async () => ({ access_token: "fresh-app-token" })); stub(github, "githubRequest", async path => { @@ -97,6 +97,9 @@ describe("registration email collection", function () { expect(saved).to.have.length(1); expect(writes).to.have.length(1); expect(identity.emails[0].email).to.equal("primary@example.com"); }); + it(`does not reset access alerts during ordinary ${provider} sign-in`, async () => { + await callback(); expect(resetCalls).to.have.length(0); + }); it(`fills missing email on existing ${provider} sign-in`, async () => { expect((await callback()).emails[0].email).to.equal("primary@example.com"); expect(saved).to.have.length(0); @@ -110,6 +113,12 @@ describe("registration email collection", function () { }); } + it("prepares recovery for an explicit OAuth reconnect", async () => { + await new Promise((resolve, reject) => verify({ githubOAuthContext: { ownerId: user.id, githubId: "10", expires: Date.now() + 60000 } }, + "token", "", { id: "10", username: "owner" }, error => error ? reject(error) : resolve())); + expect(resetCalls).to.deep.equal([user.id]); + }); + it("rejects an identity mismatch before requesting or saving email", async () => { const error = await new Promise(resolve => verify({ githubOAuthContext: { githubId: "different", expires: Date.now() + 60000 } }, "token", "", { id: "10", username: "owner" }, resolve));