mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-10 05:12:14 +00:00
Merge guardian notification delivery
This commit is contained in:
commit
2ea812d387
38 files changed
+2419
-121
No files matched your search
@@ -0,0 +1,5 @@
|
||||
-- v1.37.0 — notification event retention deletes globally by created_at.
|
||||
-- The dedup lookup index starts with record_user_id, so it cannot serve this
|
||||
-- trailing-edge cleanup predicate efficiently.
|
||||
CREATE INDEX "notification_events_created_at_idx"
|
||||
ON "notification_events" ("created_at");
|
||||
@@ -0,0 +1,41 @@
|
||||
-- v1.37.0 — a managed Guardian notification receives a durable final
|
||||
-- authorization claim under the profile lifecycle lock before provider I/O.
|
||||
-- This records only delivery principals and outcome, never health content or
|
||||
-- channel credentials. Retention follows the notification attempt cleanup.
|
||||
CREATE TABLE "notification_egress_authorizations" (
|
||||
"id" TEXT NOT NULL,
|
||||
"record_user_id" TEXT NOT NULL,
|
||||
"recipient_user_id" TEXT NOT NULL,
|
||||
"channel" TEXT NOT NULL,
|
||||
"event_type" TEXT NOT NULL,
|
||||
"authorized_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"completed_at" TIMESTAMP(3),
|
||||
"outcome" TEXT,
|
||||
|
||||
CONSTRAINT "notification_egress_authorizations_pkey" PRIMARY KEY ("id"),
|
||||
CONSTRAINT "notification_egress_authorizations_record_user_id_fkey"
|
||||
FOREIGN KEY ("record_user_id") REFERENCES "users"("id")
|
||||
ON DELETE CASCADE ON UPDATE CASCADE,
|
||||
CONSTRAINT "notification_egress_authorizations_recipient_user_id_fkey"
|
||||
FOREIGN KEY ("recipient_user_id") REFERENCES "users"("id")
|
||||
ON DELETE CASCADE ON UPDATE CASCADE
|
||||
);
|
||||
|
||||
CREATE INDEX "notification_egress_authorizations_authorized_at_idx"
|
||||
ON "notification_egress_authorizations" ("authorized_at");
|
||||
|
||||
CREATE INDEX "notification_egress_authorizations_record_user_id_recipient_user_id_authorized_at_idx"
|
||||
ON "notification_egress_authorizations" (
|
||||
"record_user_id",
|
||||
"recipient_user_id",
|
||||
"authorized_at" DESC
|
||||
);
|
||||
|
||||
-- A recipient can erase their account without owning the underlying record.
|
||||
-- PostgreSQL needs a recipient-first access path for that FK cascade and for
|
||||
-- the explicit Delete All Data predicate over recipient_user_id.
|
||||
CREATE INDEX "notification_egress_auth_recipient_authorized_at_idx"
|
||||
ON "notification_egress_authorizations" (
|
||||
"recipient_user_id",
|
||||
"authorized_at" DESC
|
||||
);
|
||||
@@ -863,6 +863,8 @@ model User {
|
||||
pushAttemptsForRecord PushAttempt[] @relation("PushAttemptRecordUser")
|
||||
pushAttemptsReceived PushAttempt[] @relation("PushAttemptRecipientUser")
|
||||
notificationEvents NotificationEvent[]
|
||||
notificationEgressAuthorizationsForRecord NotificationEgressAuthorization[] @relation("NotificationEgressAuthorizationRecordUser")
|
||||
notificationEgressAuthorizationsReceived NotificationEgressAuthorization[] @relation("NotificationEgressAuthorizationRecipientUser")
|
||||
telegramReminderMessages TelegramReminderMessage[]
|
||||
// v1.10.3 — per-(user, day) Strain personal-anchor cache (day-total
|
||||
// TRIMP + running EWMA reference) so the nightly Strain engine anchors
|
||||
@@ -5821,9 +5823,37 @@ model NotificationEvent {
|
||||
recordUser User @relation(fields: [recordUserId], references: [id], onDelete: Cascade)
|
||||
|
||||
@@index([recordUserId, eventType, dedupKey, createdAt(sort: Desc)])
|
||||
@@index([createdAt], map: "notification_events_created_at_idx")
|
||||
@@map("notification_events")
|
||||
}
|
||||
|
||||
// A managed Guardian channel send crosses a revocable authorization boundary.
|
||||
// The dispatcher creates this claim while it holds the managed-profile
|
||||
// lifecycle lock, then commits before starting provider I/O. That committed
|
||||
// claim is the linearization point: a revocation that commits first prevents
|
||||
// the send, while a revocation that commits later does not cancel the one
|
||||
// already-authorized provider operation. No notification content, channel
|
||||
// credentials, or event deduplication keys are retained here. Completion is
|
||||
// observational only; authorization and retry decisions never read it.
|
||||
model NotificationEgressAuthorization {
|
||||
id String @id @default(cuid())
|
||||
recordUserId String @map("record_user_id")
|
||||
recipientUserId String @map("recipient_user_id")
|
||||
channel String
|
||||
eventType String @map("event_type")
|
||||
authorizedAt DateTime @default(now()) @map("authorized_at")
|
||||
completedAt DateTime? @map("completed_at")
|
||||
outcome String?
|
||||
|
||||
recordUser User @relation("NotificationEgressAuthorizationRecordUser", fields: [recordUserId], references: [id], onDelete: Cascade)
|
||||
recipientUser User @relation("NotificationEgressAuthorizationRecipientUser", fields: [recipientUserId], references: [id], onDelete: Cascade)
|
||||
|
||||
@@index([authorizedAt], map: "notification_egress_authorizations_authorized_at_idx")
|
||||
@@index([recordUserId, recipientUserId, authorizedAt(sort: Desc)])
|
||||
@@index([recipientUserId, authorizedAt(sort: Desc)], map: "notification_egress_auth_recipient_authorized_at_idx")
|
||||
@@map("notification_egress_authorizations")
|
||||
}
|
||||
|
||||
// `result` is one of `"ok"` | `"error"` | `"skipped"` — `skipped`
|
||||
// covers the "no recipient" / "config missing" soft cases.
|
||||
//
|
||||
|
||||
@@ -146,6 +146,15 @@ describe("every schema model has a backup verdict", () => {
|
||||
"a one-word reason is how a decision stops being reviewable",
|
||||
).toEqual([]);
|
||||
});
|
||||
|
||||
it("excludes Guardian egress claims rather than exporting delivery metadata", () => {
|
||||
expect(backupVerdict("NotificationEgressAuthorization")).toBe(
|
||||
"NOT_IN_BACKUP",
|
||||
);
|
||||
expect(NOT_IN_BACKUP_MODELS.NotificationEgressAuthorization).toMatch(
|
||||
/no notification content, credentials, or deduplication key/i,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
|
||||
@@ -341,4 +341,10 @@ describe("data-wipe completeness", () => {
|
||||
});
|
||||
expect(wipeWhere("Measurement", "u1")).toEqual({ userId: "u1" });
|
||||
});
|
||||
|
||||
it("selects both record and recipient ownership of a Guardian egress claim", () => {
|
||||
expect(wipeWhere("NotificationEgressAuthorization", "u1")).toEqual({
|
||||
OR: [{ recordUserId: "u1" }, { recipientUserId: "u1" }],
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,21 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import { isManagedGuardianDelivery } from "@/lib/notifications/managed-delivery";
|
||||
|
||||
describe("managed Guardian notification payload safety", () => {
|
||||
it("recognises only cross-account record delivery as managed Guardian delivery", () => {
|
||||
expect(
|
||||
isManagedGuardianDelivery({
|
||||
recordUserId: "record-1",
|
||||
recipientUserId: "guardian-1",
|
||||
}),
|
||||
).toBe(true);
|
||||
|
||||
expect(
|
||||
isManagedGuardianDelivery({
|
||||
recordUserId: "user-1",
|
||||
recipientUserId: "user-1",
|
||||
}),
|
||||
).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,118 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const { findUnique, findMany } = vi.hoisted(() => ({
|
||||
findUnique: vi.fn(),
|
||||
findMany: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
user: { findUnique },
|
||||
accountGrant: { findMany },
|
||||
},
|
||||
}));
|
||||
|
||||
import {
|
||||
isManagedGuardianFanoutAllowed,
|
||||
resolveManagedGuardianRecipientIds,
|
||||
} from "@/lib/notifications/delivery-identity";
|
||||
|
||||
const recordUserId = "managed-record";
|
||||
|
||||
function payload(over: Record<string, unknown> = {}) {
|
||||
return {
|
||||
eventType: "MEDICATION_REMINDER" as const,
|
||||
userId: recordUserId,
|
||||
title: "Record content",
|
||||
message: "Record schedule",
|
||||
...over,
|
||||
};
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.resetAllMocks();
|
||||
findUnique.mockResolvedValue({ managedProfileAt: new Date() });
|
||||
findMany.mockResolvedValue([
|
||||
{ granteeId: "guardian-one", grantee: { managedProfileAt: null } },
|
||||
]);
|
||||
});
|
||||
|
||||
describe("guardian notification policy — allowlist and eligibility", () => {
|
||||
it("allows exactly medication, measurement, low-stock, and explicitly tagged safety-floor alerts", () => {
|
||||
expect(isManagedGuardianFanoutAllowed(payload())).toBe(true);
|
||||
expect(
|
||||
isManagedGuardianFanoutAllowed(
|
||||
payload({ eventType: "MEASUREMENT_REMINDER" }),
|
||||
),
|
||||
).toBe(true);
|
||||
expect(
|
||||
isManagedGuardianFanoutAllowed(
|
||||
payload({ eventType: "MEDICATION_LOW_STOCK" }),
|
||||
),
|
||||
).toBe(true);
|
||||
expect(
|
||||
isManagedGuardianFanoutAllowed(
|
||||
payload({
|
||||
eventType: "SYSTEM_ALERT",
|
||||
managedFanoutEvent: "SAFETY_FLOOR_ALERT",
|
||||
}),
|
||||
),
|
||||
).toBe(true);
|
||||
|
||||
expect(
|
||||
isManagedGuardianFanoutAllowed(payload({ eventType: "SYSTEM_ALERT" })),
|
||||
).toBe(false);
|
||||
expect(
|
||||
isManagedGuardianFanoutAllowed(
|
||||
payload({ eventType: "MEASUREMENT_ANOMALY" }),
|
||||
),
|
||||
).toBe(false);
|
||||
expect(
|
||||
isManagedGuardianFanoutAllowed(
|
||||
payload({
|
||||
eventType: "SYSTEM_ALERT",
|
||||
managedFanoutEvent: "ILLNESS_RED_FLAG",
|
||||
}),
|
||||
),
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
it("enumerates only active Guardians of the marked record", async () => {
|
||||
await expect(
|
||||
resolveManagedGuardianRecipientIds(payload()),
|
||||
).resolves.toEqual(["guardian-one"]);
|
||||
|
||||
expect(findMany).toHaveBeenCalledWith({
|
||||
where: expect.objectContaining({ grantorId: recordUserId }),
|
||||
select: {
|
||||
granteeId: true,
|
||||
grantee: { select: { managedProfileAt: true } },
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it("does not fan out unmarked records, disallowed events, self delivery, or invalid Guardians", async () => {
|
||||
findUnique.mockResolvedValueOnce({ managedProfileAt: null });
|
||||
await expect(
|
||||
resolveManagedGuardianRecipientIds(payload()),
|
||||
).resolves.toBeNull();
|
||||
|
||||
await expect(
|
||||
resolveManagedGuardianRecipientIds(
|
||||
payload({ eventType: "MEASUREMENT_ANOMALY" }),
|
||||
),
|
||||
).resolves.toBeNull();
|
||||
|
||||
findMany.mockResolvedValueOnce([
|
||||
{ granteeId: recordUserId, grantee: { managedProfileAt: null } },
|
||||
{
|
||||
granteeId: "managed-guardian",
|
||||
grantee: { managedProfileAt: new Date() },
|
||||
},
|
||||
{ granteeId: "guardian-two", grantee: { managedProfileAt: null } },
|
||||
]);
|
||||
await expect(
|
||||
resolveManagedGuardianRecipientIds(payload()),
|
||||
).resolves.toEqual(["guardian-two"]);
|
||||
});
|
||||
});
|
||||
@@ -1,6 +1,8 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const transactionMock = vi.fn();
|
||||
const authorizationClaimQueryMock = vi.fn();
|
||||
const authorizationUpdateMock = vi.fn();
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
@@ -10,6 +12,10 @@ vi.mock("@/lib/db", () => ({
|
||||
upsert: vi.fn(),
|
||||
},
|
||||
user: { findUnique: vi.fn() },
|
||||
accountGrant: { findMany: vi.fn() },
|
||||
notificationEgressAuthorization: {
|
||||
update: (...args: unknown[]) => authorizationUpdateMock(...args),
|
||||
},
|
||||
auditLog: { create: vi.fn() },
|
||||
$transaction: (...args: unknown[]) => transactionMock(...args),
|
||||
},
|
||||
@@ -32,15 +38,17 @@ vi.mock("@/lib/logging/context", () => ({
|
||||
}));
|
||||
|
||||
const sendViaWebPushMock = vi.fn();
|
||||
const sendViaNtfyMock = vi.fn();
|
||||
const sendViaApnsMock = vi.fn();
|
||||
|
||||
vi.mock("@/lib/notifications/senders/apns", () => ({
|
||||
sendViaApns: vi.fn(),
|
||||
sendViaApns: (...args: unknown[]) => sendViaApnsMock(...args),
|
||||
}));
|
||||
vi.mock("@/lib/notifications/senders/email", () => ({
|
||||
sendViaEmail: vi.fn(),
|
||||
}));
|
||||
vi.mock("@/lib/notifications/senders/ntfy", () => ({
|
||||
sendViaNtfy: vi.fn(),
|
||||
sendViaNtfy: (...args: unknown[]) => sendViaNtfyMock(...args),
|
||||
}));
|
||||
vi.mock("@/lib/notifications/senders/telegram", () => ({
|
||||
sendViaTelegram: vi.fn(),
|
||||
@@ -63,6 +71,8 @@ const recipientUserId = "guardian-recipient";
|
||||
|
||||
beforeEach(() => {
|
||||
vi.resetAllMocks();
|
||||
authorizationClaimQueryMock.mockResolvedValue([{ id: "authorization" }]);
|
||||
authorizationUpdateMock.mockResolvedValue({ id: "authorization" });
|
||||
vi.mocked(prisma.user.findUnique).mockResolvedValue({
|
||||
managedProfileAt: new Date(),
|
||||
} as never);
|
||||
@@ -82,7 +92,7 @@ beforeEach(() => {
|
||||
transactionMock.mockImplementation(
|
||||
async (operation: (tx: unknown) => Promise<unknown>) =>
|
||||
operation({
|
||||
$queryRaw: vi.fn(),
|
||||
$queryRaw: (...args: unknown[]) => authorizationClaimQueryMock(...args),
|
||||
user: {
|
||||
findUnique: vi.fn(async ({ where }: { where: { id: string } }) =>
|
||||
where.id === recordUserId
|
||||
@@ -94,6 +104,8 @@ beforeEach(() => {
|
||||
}),
|
||||
);
|
||||
sendViaWebPushMock.mockResolvedValue({ ok: true });
|
||||
sendViaNtfyMock.mockResolvedValue({ ok: true });
|
||||
sendViaApnsMock.mockResolvedValue({ ok: true });
|
||||
});
|
||||
|
||||
describe("notification delivery identity", () => {
|
||||
@@ -127,7 +139,80 @@ describe("notification delivery identity", () => {
|
||||
message: "Record schedule",
|
||||
}),
|
||||
);
|
||||
expect(transactionMock).toHaveBeenCalledTimes(1);
|
||||
// Preflight chooses recipient channels, then a short locked transaction
|
||||
// records the final authorization claim before provider I/O starts.
|
||||
expect(transactionMock).toHaveBeenCalledTimes(2);
|
||||
expect(transactionMock).toHaveBeenLastCalledWith(expect.any(Function));
|
||||
expect(authorizationClaimQueryMock).toHaveBeenCalledTimes(3);
|
||||
expect(authorizationUpdateMock).toHaveBeenCalledWith({
|
||||
where: { id: "authorization" },
|
||||
data: { completedAt: expect.any(Date), outcome: "ok" },
|
||||
});
|
||||
});
|
||||
|
||||
it("does not invoke a provider when final Guardian authorization is revoked", async () => {
|
||||
let transactionCount = 0;
|
||||
transactionMock.mockImplementation(
|
||||
async (operation: (tx: unknown) => Promise<unknown>) => {
|
||||
transactionCount += 1;
|
||||
return operation({
|
||||
$queryRaw: (...args: unknown[]) =>
|
||||
authorizationClaimQueryMock(...args),
|
||||
user: {
|
||||
findUnique: vi.fn(async ({ where }: { where: { id: string } }) =>
|
||||
where.id === recordUserId
|
||||
? { managedProfileAt: new Date() }
|
||||
: { managedProfileAt: null },
|
||||
),
|
||||
},
|
||||
accountGrant: {
|
||||
findFirst: vi.fn(async () =>
|
||||
transactionCount === 1 ? { id: "grant" } : null,
|
||||
),
|
||||
},
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
const outcome = await dispatchNotification({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: recordUserId,
|
||||
recordUserId,
|
||||
recipientUserId,
|
||||
title: "Record content",
|
||||
message: "Record schedule",
|
||||
});
|
||||
|
||||
expect(outcome).toEqual({
|
||||
dispatched: false,
|
||||
channelsAttempted: 0,
|
||||
channelsSucceeded: 0,
|
||||
});
|
||||
expect(authorizationClaimQueryMock).toHaveBeenCalledTimes(2);
|
||||
expect(sendViaWebPushMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("records a provider timeout on the claim without retaining the lifecycle lock", async () => {
|
||||
sendViaWebPushMock.mockRejectedValue(new Error("provider timeout"));
|
||||
|
||||
const outcome = await dispatchNotification({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: recordUserId,
|
||||
recordUserId,
|
||||
recipientUserId,
|
||||
title: "Record content",
|
||||
message: "Record schedule",
|
||||
});
|
||||
|
||||
expect(outcome).toEqual({
|
||||
dispatched: false,
|
||||
channelsAttempted: 1,
|
||||
channelsSucceeded: 0,
|
||||
});
|
||||
expect(authorizationUpdateMock).toHaveBeenCalledWith({
|
||||
where: { id: "authorization" },
|
||||
data: { completedAt: expect.any(Date), outcome: "sender_threw" },
|
||||
});
|
||||
});
|
||||
|
||||
it("refuses a managed record without an explicit recipient", async () => {
|
||||
@@ -180,4 +265,155 @@ describe("notification delivery identity", () => {
|
||||
);
|
||||
expect(sendViaWebPushMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("renders for each Guardian while isolating channel preferences", async () => {
|
||||
const firstGuardian = "guardian-en";
|
||||
const secondGuardian = "guardian-de";
|
||||
|
||||
(
|
||||
prisma.user.findUnique as unknown as ReturnType<typeof vi.fn>
|
||||
).mockImplementation(async ({ where }: { where: { id: string } }) => {
|
||||
if (where.id === recordUserId) {
|
||||
return { managedProfileAt: new Date() } as never;
|
||||
}
|
||||
if (where.id === firstGuardian) {
|
||||
return {
|
||||
managedProfileAt: null,
|
||||
locale: "en",
|
||||
notificationPrefs: { medication: { clientManaged: true } },
|
||||
} as never;
|
||||
}
|
||||
return {
|
||||
managedProfileAt: null,
|
||||
locale: "de",
|
||||
notificationPrefs: { medication: { clientManaged: true } },
|
||||
} as never;
|
||||
});
|
||||
vi.mocked(prisma.accountGrant.findMany).mockResolvedValue([
|
||||
{ granteeId: firstGuardian, grantee: { managedProfileAt: null } },
|
||||
{ granteeId: secondGuardian, grantee: { managedProfileAt: null } },
|
||||
] as never);
|
||||
(
|
||||
prisma.notificationChannel.findMany as unknown as ReturnType<typeof vi.fn>
|
||||
).mockImplementation(async ({ where }: { where: { userId: string } }) => {
|
||||
if (where.userId === firstGuardian) {
|
||||
return [
|
||||
{
|
||||
id: "first-web-disabled",
|
||||
userId: firstGuardian,
|
||||
type: "WEB_PUSH",
|
||||
config: "{}",
|
||||
nextRetryAt: null,
|
||||
preferences: [{ eventType: "MEDICATION_REMINDER", enabled: false }],
|
||||
},
|
||||
{
|
||||
id: "first-ntfy-enabled",
|
||||
userId: firstGuardian,
|
||||
type: "NTFY",
|
||||
config: "{}",
|
||||
nextRetryAt: null,
|
||||
preferences: [],
|
||||
},
|
||||
] as never;
|
||||
}
|
||||
return [
|
||||
{
|
||||
id: "second-web-enabled",
|
||||
userId: secondGuardian,
|
||||
type: "WEB_PUSH",
|
||||
config: "{}",
|
||||
nextRetryAt: null,
|
||||
preferences: [],
|
||||
},
|
||||
{
|
||||
id: "second-ntfy-disabled",
|
||||
userId: secondGuardian,
|
||||
type: "NTFY",
|
||||
config: "{}",
|
||||
nextRetryAt: null,
|
||||
preferences: [{ eventType: "MEDICATION_REMINDER", enabled: false }],
|
||||
},
|
||||
] as never;
|
||||
});
|
||||
|
||||
const outcome = await dispatchNotification({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: recordUserId,
|
||||
title: "record-language-title",
|
||||
message: "record-language-message",
|
||||
renderForRecipient: (locale) => ({
|
||||
title: `title-${locale}`,
|
||||
message: `message-${locale}`,
|
||||
}),
|
||||
});
|
||||
|
||||
expect(outcome).toEqual({
|
||||
dispatched: true,
|
||||
channelsAttempted: 2,
|
||||
channelsSucceeded: 2,
|
||||
});
|
||||
expect(sendViaNtfyMock).toHaveBeenCalledWith(
|
||||
{},
|
||||
expect.objectContaining({
|
||||
recipientUserId: firstGuardian,
|
||||
title: "title-en",
|
||||
message: "message-en",
|
||||
}),
|
||||
);
|
||||
expect(sendViaWebPushMock).toHaveBeenCalledWith(
|
||||
secondGuardian,
|
||||
expect.objectContaining({
|
||||
recipientUserId: secondGuardian,
|
||||
title: "title-de",
|
||||
message: "message-de",
|
||||
}),
|
||||
);
|
||||
expect(sendViaWebPushMock).toHaveBeenCalledTimes(1);
|
||||
expect(sendViaNtfyMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("bypasses client-managed APNs suppression only for managed Guardian delivery", async () => {
|
||||
const guardian = "guardian-apns";
|
||||
(
|
||||
prisma.user.findUnique as unknown as ReturnType<typeof vi.fn>
|
||||
).mockImplementation(
|
||||
async ({ where }: { where: { id: string } }) =>
|
||||
(where.id === recordUserId
|
||||
? { managedProfileAt: new Date() }
|
||||
: {
|
||||
managedProfileAt: null,
|
||||
notificationPrefs: { medication: { clientManaged: true } },
|
||||
}) as never,
|
||||
);
|
||||
vi.mocked(prisma.accountGrant.findMany).mockResolvedValue([
|
||||
{ granteeId: guardian, grantee: { managedProfileAt: null } },
|
||||
] as never);
|
||||
vi.mocked(prisma.notificationChannel.findMany).mockResolvedValue([
|
||||
{
|
||||
id: "guardian-apns",
|
||||
userId: guardian,
|
||||
type: "APNS",
|
||||
config: "{}",
|
||||
nextRetryAt: null,
|
||||
preferences: [],
|
||||
},
|
||||
] as never);
|
||||
|
||||
await expect(
|
||||
dispatchNotification({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: recordUserId,
|
||||
title: "Record content",
|
||||
message: "Record schedule",
|
||||
}),
|
||||
).resolves.toMatchObject({ dispatched: true, channelsSucceeded: 1 });
|
||||
|
||||
expect(sendViaApnsMock).toHaveBeenCalledWith(
|
||||
guardian,
|
||||
expect.objectContaining({
|
||||
recordUserId,
|
||||
recipientUserId: guardian,
|
||||
}),
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -9,6 +9,10 @@ import {
|
||||
} from "@/lib/api-handler";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import { invalidateUserData } from "@/lib/cache/invalidate";
|
||||
import {
|
||||
LastManagedGuardianError,
|
||||
protectManagedProfilesDuringDataWipe,
|
||||
} from "@/lib/managed-profiles/lifecycle";
|
||||
import {
|
||||
resolveWipeDelegate,
|
||||
USER_RESET,
|
||||
@@ -66,27 +70,45 @@ export const DELETE = apiHandler(async (request: NextRequest) => {
|
||||
|
||||
const userId = user.id;
|
||||
|
||||
const counts = await prisma.$transaction(
|
||||
async (tx) => {
|
||||
const perModel: Record<string, number> = {};
|
||||
for (const model of WIPE_MODELS) {
|
||||
const { count } = await resolveWipeDelegate(tx, model).deleteMany({
|
||||
where: { userId },
|
||||
});
|
||||
if (count > 0) perModel[wipeDelegateKey(model)] = count;
|
||||
}
|
||||
let counts: Record<string, number>;
|
||||
try {
|
||||
counts = await prisma.$transaction(
|
||||
async (tx) => {
|
||||
// This precedes every data delete and holds the managed-record locks
|
||||
// through AccountGrant removal. A queued Guardian egress claim thus
|
||||
// rechecks after this wipe commits instead of sending from a grant
|
||||
// this transaction has already deleted.
|
||||
await protectManagedProfilesDuringDataWipe(tx, userId);
|
||||
|
||||
// The account row survives; the personal data carried on its own columns
|
||||
// does not. Same classification contract as the model list.
|
||||
await tx.user.update({ where: { id: userId }, data: USER_RESET });
|
||||
const perModel: Record<string, number> = {};
|
||||
for (const model of WIPE_MODELS) {
|
||||
const { count } = await resolveWipeDelegate(tx, model).deleteMany({
|
||||
where: { userId },
|
||||
});
|
||||
if (count > 0) perModel[wipeDelegateKey(model)] = count;
|
||||
}
|
||||
|
||||
return perModel;
|
||||
},
|
||||
{
|
||||
timeout: WIPE_TRANSACTION_TIMEOUT_MS,
|
||||
maxWait: WIPE_TRANSACTION_MAX_WAIT_MS,
|
||||
},
|
||||
);
|
||||
// The account row survives; the personal data carried on its own columns
|
||||
// does not. Same classification contract as the model list.
|
||||
await tx.user.update({ where: { id: userId }, data: USER_RESET });
|
||||
|
||||
return perModel;
|
||||
},
|
||||
{
|
||||
timeout: WIPE_TRANSACTION_TIMEOUT_MS,
|
||||
maxWait: WIPE_TRANSACTION_MAX_WAIT_MS,
|
||||
},
|
||||
);
|
||||
} catch (error) {
|
||||
if (error instanceof LastManagedGuardianError) {
|
||||
return apiError(
|
||||
"Add another Guardian or delete the managed profile first",
|
||||
409,
|
||||
{ errorCode: "managed_profile.guardian.required" },
|
||||
);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
||||
const deletedRows = Object.values(counts).reduce((sum, n) => sum + n, 0);
|
||||
|
||||
|
||||
@@ -287,6 +287,25 @@ describe("Telegram webhook — rate limit", () => {
|
||||
});
|
||||
|
||||
describe("Telegram webhook — callback dispatch", () => {
|
||||
it("rejects a callback tied to a managed Guardian reminder without mutating the record", async () => {
|
||||
vi.mocked(prisma.telegramReminderMessage.findFirst).mockResolvedValueOnce({
|
||||
medication: { userId: "managed-record" },
|
||||
} as never);
|
||||
|
||||
const res = await POST(tgRequest(callbackUpdate("taken:med-1")));
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(prisma.medication.findFirst).not.toHaveBeenCalled();
|
||||
expect(prisma.medicationIntakeEvent.create).not.toHaveBeenCalled();
|
||||
expect(prisma.medication.update).not.toHaveBeenCalled();
|
||||
expect(deleteMessage).not.toHaveBeenCalled();
|
||||
expect(answerTelegramCallbackQuery).toHaveBeenCalledWith(
|
||||
"decrypted:ENC(token-blob)",
|
||||
"cb-1",
|
||||
"Invalid action.",
|
||||
);
|
||||
});
|
||||
|
||||
it("'taken:<medId>' creates a MedicationIntakeEvent + clears snoozedUntil + acks the callback", async () => {
|
||||
const res = await POST(tgRequest(callbackUpdate("taken:med-1")));
|
||||
expect(res.status).toBe(200);
|
||||
|
||||
@@ -162,6 +162,10 @@ export const WIPE_MODELS = [
|
||||
"PushSubscription",
|
||||
"PushAttempt",
|
||||
"NotificationEvent",
|
||||
// A Guardian egress claim belongs to both the record whose data prompted
|
||||
// the notification and the recipient whose channel would receive it. A
|
||||
// wipe by either account must remove it; see WIPE_OWNER_FIELDS below.
|
||||
"NotificationEgressAuthorization",
|
||||
"Device",
|
||||
"TelegramScheduledDeletion",
|
||||
"TelegramPromptContext",
|
||||
@@ -423,6 +427,7 @@ export const WIPE_OWNER_FIELDS: Readonly<Record<string, readonly string[]>> = {
|
||||
// import history.
|
||||
MedicationIntakeImportJob: ["recordUserId"],
|
||||
NotificationEvent: ["recordUserId"],
|
||||
NotificationEgressAuthorization: ["recordUserId", "recipientUserId"],
|
||||
// Both sides. Wiping only the grantor side would leave the account still
|
||||
// holding read access to other people's records after it asked for
|
||||
// everything of its own to be deleted.
|
||||
|
||||
@@ -408,6 +408,8 @@ export const NOT_IN_BACKUP_MODELS: Readonly<Record<string, string>> = {
|
||||
"A 90-day delivery ledger for this host's sends. Restoring it onto another host would assert sends it never made.",
|
||||
NotificationEvent:
|
||||
"A local notification dedup anchor. Restoring it onto another host would suppress reminders and safety notices that host still owes.",
|
||||
NotificationEgressAuthorization:
|
||||
"A short-lived, deployment-local Guardian egress ledger. It retains no notification content, credentials, or deduplication key, and its outcome is observation rather than authorization; restoring it would nevertheless carry stale send history to a host that must recheck the live grant before every send. Exclude it so a backup never exports notification-derived privacy metadata or makes a restored host reason from another host's delivery work.",
|
||||
AuditLog:
|
||||
"An append-only record of actions taken on this instance. Merging one instance's audit trail into another's would make the trail untrue.",
|
||||
CoachUsage:
|
||||
|
||||
@@ -124,15 +124,24 @@ describe("notifySafetyFloor", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("does not dispatch or write an anchor for a managed profile", async () => {
|
||||
it("claims then admits an anchored Guardian fan-out for a managed profile", async () => {
|
||||
userFindUniqueMock.mockResolvedValueOnce({
|
||||
managedProfileAt: new Date("2026-08-06T00:00:00.000Z"),
|
||||
});
|
||||
|
||||
await notifySafetyFloor({ userId: "managed-record", decision: bpHigh });
|
||||
|
||||
expect(claimMock).not.toHaveBeenCalled();
|
||||
expect(dispatchMock).not.toHaveBeenCalled();
|
||||
expect(claimMock).toHaveBeenCalledWith(
|
||||
expect.anything(),
|
||||
expect.objectContaining({ recordUserId: "managed-record" }),
|
||||
);
|
||||
expect(dispatchMock).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
userId: "managed-record",
|
||||
eventType: "SYSTEM_ALERT",
|
||||
managedFanoutEvent: "SAFETY_FLOOR_ALERT",
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("never throws when dispatch fails", async () => {
|
||||
|
||||
@@ -45,8 +45,9 @@ export async function notifyIllnessRedFlag(input: {
|
||||
if (redFlags.length === 0) return;
|
||||
|
||||
try {
|
||||
// Plan 11 adds approved guardian fanout for this safety signal. Until
|
||||
// then, managed profiles must not produce a direct delivery or anchor.
|
||||
// Red-flag alerts are deliberately outside the closed Guardian fan-out
|
||||
// allowlist. Managed records therefore retain the existing no-delivery,
|
||||
// no-anchor behaviour for this distinct SYSTEM_ALERT producer.
|
||||
const user = await prisma.user.findUnique({
|
||||
where: { id: userId },
|
||||
select: { managedProfileAt: true },
|
||||
|
||||
@@ -114,14 +114,6 @@ export async function notifySafetyFloor(input: {
|
||||
if (!decision) return;
|
||||
|
||||
try {
|
||||
// Plan 11 adds approved guardian fanout for this safety signal. Until
|
||||
// then, managed profiles must not produce a direct delivery or anchor.
|
||||
const user = await prisma.user.findUnique({
|
||||
where: { id: userId },
|
||||
select: { managedProfileAt: true },
|
||||
});
|
||||
if (user?.managedProfileAt) return;
|
||||
|
||||
const reason = `${LEDGER_REASON_PREFIX}${decision.reason}`;
|
||||
const since = new Date(Date.now() - DEDUPE_WINDOW_MS);
|
||||
const claimed = await claimNotificationEvent(prisma, {
|
||||
@@ -139,6 +131,7 @@ export async function notifySafetyFloor(input: {
|
||||
messageKey,
|
||||
params: paramsFor(decision, input.glucoseUnit ?? "mg/dL"),
|
||||
eventType: "SYSTEM_ALERT",
|
||||
managedFanoutEvent: "SAFETY_FLOOR_ALERT",
|
||||
urgent: true,
|
||||
});
|
||||
|
||||
|
||||
@@ -329,12 +329,21 @@ export async function runMeasurementReminderTick(
|
||||
reminder.label,
|
||||
reminder.location,
|
||||
);
|
||||
const renderForRecipient = (locale: Locale) => {
|
||||
const rendered = buildMeasurementReminderPayload(
|
||||
locale,
|
||||
reminder.label,
|
||||
reminder.location,
|
||||
);
|
||||
return { title: rendered.title, message: rendered.body };
|
||||
};
|
||||
|
||||
const outcome = await dispatchImpl({
|
||||
eventType: "MEASUREMENT_REMINDER",
|
||||
userId: reminder.user.id,
|
||||
title,
|
||||
message: body,
|
||||
renderForRecipient,
|
||||
metadata: {
|
||||
scheduledAt: now.toISOString(),
|
||||
reminderId: reminder.id,
|
||||
|
||||
@@ -497,11 +497,26 @@ export async function runMedicationLowStockTick(
|
||||
schedules: med.schedules,
|
||||
today: now,
|
||||
});
|
||||
const renderForRecipient = (locale: Locale) => {
|
||||
const rendered = buildLowStockPayload({
|
||||
locale,
|
||||
medName: med.name,
|
||||
runwayDays: evaluation.runwayDays ?? 0,
|
||||
unitsRemaining: evaluation.unitsRemaining,
|
||||
expiredUnits: evaluation.expiredUnits,
|
||||
leadDays,
|
||||
triggerDays,
|
||||
schedules: med.schedules,
|
||||
today: now,
|
||||
});
|
||||
return { title: rendered.title, message: rendered.body };
|
||||
};
|
||||
const outcome = await dispatchImpl({
|
||||
eventType: "MEDICATION_LOW_STOCK",
|
||||
userId: user.id,
|
||||
title,
|
||||
message: body,
|
||||
renderForRecipient,
|
||||
metadata: {
|
||||
scheduledAt: now.toISOString(),
|
||||
medicationId: med.id,
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const h = vi.hoisted(() => ({
|
||||
pushDeleteMany: vi.fn(),
|
||||
eventDeleteMany: vi.fn(),
|
||||
authorizationDeleteMany: vi.fn(),
|
||||
meta: {} as Record<string, unknown>,
|
||||
}));
|
||||
|
||||
vi.mock("../shared", async (importOriginal) => ({
|
||||
...(await importOriginal<Record<string, unknown>>()),
|
||||
getWorkerPrisma: () => ({
|
||||
pushAttempt: { deleteMany: h.pushDeleteMany },
|
||||
notificationEvent: { deleteMany: h.eventDeleteMany },
|
||||
notificationEgressAuthorization: { deleteMany: h.authorizationDeleteMany },
|
||||
}),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/logging/background", () => ({
|
||||
withBackgroundEvent: async (
|
||||
_name: string,
|
||||
fn: (evt: {
|
||||
addMeta: (key: string, value: unknown) => void;
|
||||
addWarning: (warning: string) => void;
|
||||
}) => Promise<unknown>,
|
||||
) =>
|
||||
fn({
|
||||
addMeta: (key, value) => {
|
||||
h.meta[key] = value;
|
||||
},
|
||||
addWarning: vi.fn(),
|
||||
}),
|
||||
}));
|
||||
|
||||
import { handlePushAttemptCleanup } from "../cleanup-handlers";
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
for (const key of Object.keys(h.meta)) delete h.meta[key];
|
||||
h.pushDeleteMany.mockResolvedValue({ count: 3 });
|
||||
h.eventDeleteMany.mockResolvedValue({ count: 5 });
|
||||
h.authorizationDeleteMany.mockResolvedValue({ count: 7 });
|
||||
});
|
||||
|
||||
describe("notification retention", () => {
|
||||
it("prunes notification ledgers with the canonical bounded attempt retention", async () => {
|
||||
await handlePushAttemptCleanup([]);
|
||||
|
||||
expect(h.pushDeleteMany).toHaveBeenCalledWith({
|
||||
where: { createdAt: { lt: expect.any(Date) } },
|
||||
});
|
||||
expect(h.eventDeleteMany).toHaveBeenCalledWith({
|
||||
where: { createdAt: { lt: expect.any(Date) } },
|
||||
});
|
||||
expect(h.authorizationDeleteMany).toHaveBeenCalledWith({
|
||||
where: { authorizedAt: { lt: expect.any(Date) } },
|
||||
});
|
||||
expect(h.meta.push_attempt_cleanup_deleted).toBe(3);
|
||||
expect(h.meta.notification_event_cleanup_deleted).toBe(5);
|
||||
expect(h.meta.notification_egress_authorization_cleanup_deleted).toBe(7);
|
||||
});
|
||||
});
|
||||
@@ -41,8 +41,9 @@ import {
|
||||
import { getWorkerPrisma } from "./shared";
|
||||
|
||||
const MOOD_REMINDER_RETENTION_DAYS = 90;
|
||||
// v1.4.49 — daily prune for the push-attempt ledger. Same 90-day
|
||||
// retention as the mood-reminder dispatch ledger; both surfaces are
|
||||
// v1.4.49 — daily prune for the notification delivery ledger. The attempt
|
||||
// diagnostics and record-only event anchors share the same 90-day horizon;
|
||||
// both surfaces are
|
||||
// behavioural footprints we keep long enough to debug a duplicate-push
|
||||
// report (~one billing cycle) but no longer. Slots at 03:35 between
|
||||
// mood-reminder cleanup (03:25) and drain-cumulative (03:45) so the
|
||||
@@ -116,18 +117,20 @@ export async function handleMoodReminderCleanup(
|
||||
}
|
||||
|
||||
/**
|
||||
* v1.4.49 — daily prune for the per-attempt push-delivery ledger.
|
||||
* Daily prune for the bounded notification-delivery ledger.
|
||||
*
|
||||
* Every sender (APNS, WEB_PUSH, TELEGRAM, NTFY) writes one
|
||||
* fire-and-forget row to `push_attempts` per dispatch. The admin
|
||||
* diagnostic endpoint only ever reads the trailing 20 rows per user,
|
||||
* so anything older than the 90-day retention window is dead weight
|
||||
* inflating the table and the `(user_id, created_at DESC)` index.
|
||||
* inflating the table and the `(user_id, created_at DESC)` index. Record-only
|
||||
* notification events are claims rather than delivery diagnostics, but after
|
||||
* the same horizon their dedup windows have elapsed and they have no reader.
|
||||
*
|
||||
* The DELETE is unbounded by user — the index covers `created_at`
|
||||
* directly, so a `WHERE created_at < cutoff` scan is bounded by the
|
||||
* size of the trailing-edge of the table rather than the live working
|
||||
* set. On a one-million-row table with the documented retention
|
||||
* The DELETE is unbounded by user. The event and attempt ledgers carry direct
|
||||
* `created_at` indexes, and final egress claims carry `authorized_at`, so each
|
||||
* trailing-edge scan is bounded by the size of the live working set. On a
|
||||
* one-million-row table with the documented retention
|
||||
* window, the daily prune touches ~11k rows (1M / 90d × 1d) and
|
||||
* completes in milliseconds.
|
||||
*/
|
||||
@@ -144,11 +147,27 @@ export async function handlePushAttemptCleanup(
|
||||
try {
|
||||
const cutoff = new Date();
|
||||
cutoff.setUTCDate(cutoff.getUTCDate() - PUSH_ATTEMPT_RETENTION_DAYS);
|
||||
const deleted = await p.pushAttempt.deleteMany({
|
||||
const deletedAttempts = await p.pushAttempt.deleteMany({
|
||||
where: { createdAt: { lt: cutoff } },
|
||||
});
|
||||
evt.addMeta("push_attempt_cleanup_deleted", deleted.count);
|
||||
return jobDone({ deleted: deleted.count });
|
||||
const deletedEvents = await p.notificationEvent.deleteMany({
|
||||
where: { createdAt: { lt: cutoff } },
|
||||
});
|
||||
const deletedAuthorizations =
|
||||
await p.notificationEgressAuthorization.deleteMany({
|
||||
where: { authorizedAt: { lt: cutoff } },
|
||||
});
|
||||
const deleted =
|
||||
deletedAttempts.count +
|
||||
deletedEvents.count +
|
||||
deletedAuthorizations.count;
|
||||
evt.addMeta("push_attempt_cleanup_deleted", deletedAttempts.count);
|
||||
evt.addMeta("notification_event_cleanup_deleted", deletedEvents.count);
|
||||
evt.addMeta(
|
||||
"notification_egress_authorization_cleanup_deleted",
|
||||
deletedAuthorizations.count,
|
||||
);
|
||||
return jobDone({ deleted });
|
||||
} catch (err) {
|
||||
evt.addWarning(`push-attempt-cleanup failed: ${err}`);
|
||||
return jobFailed("push-attempt cleanup failed", err);
|
||||
|
||||
@@ -5,6 +5,7 @@
|
||||
* schedules, and boss.work registrations.
|
||||
*/
|
||||
import { type Job } from "pg-boss";
|
||||
import type { Locale } from "@/lib/i18n/config";
|
||||
import { recordError, recordReminderCheck } from "@/lib/jobs/worker-status";
|
||||
import { recomputeMedicationComplianceForEvent } from "@/lib/rollups/medication-compliance-rollups";
|
||||
import { decrypt } from "@/lib/crypto";
|
||||
@@ -630,6 +631,15 @@ export async function handleReminderCheck(
|
||||
Math.ceil(minutesToEnd),
|
||||
med.user.locale,
|
||||
);
|
||||
const renderForRecipient = (locale: Locale) =>
|
||||
getPhaseMessage(
|
||||
currentPhase,
|
||||
med.name,
|
||||
doseInfo,
|
||||
timeWindow,
|
||||
Math.ceil(minutesToEnd),
|
||||
locale,
|
||||
);
|
||||
|
||||
const keyboard = getPhaseKeyboard(
|
||||
currentPhase,
|
||||
@@ -652,6 +662,7 @@ export async function handleReminderCheck(
|
||||
userId: med.user.id,
|
||||
title,
|
||||
message,
|
||||
renderForRecipient,
|
||||
metadata: {
|
||||
medicationId: med.id,
|
||||
scheduleId: schedule.id,
|
||||
|
||||
@@ -121,6 +121,68 @@ export async function deleteGuardianAccountWithLifecycle<T>(
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Hold every managed-record lock affected by a data wipe, then prove that the
|
||||
* wipe will not leave a managed profile with no active Guardian. The caller
|
||||
* keeps the transaction open through the AccountGrant deletion, so a final
|
||||
* notification egress claim observes either the live grants before the wipe
|
||||
* or the committed deletion afterwards; it cannot authorize in between.
|
||||
*
|
||||
* A data wipe keeps the account and its managed-profile marker. It therefore
|
||||
* refuses rather than silently turning the last Guardian removal into an
|
||||
* unmanaged profile. Deleting the managed profile remains the explicit
|
||||
* product action for that case.
|
||||
*/
|
||||
export async function protectManagedProfilesDuringDataWipe(
|
||||
tx: Transaction,
|
||||
userId: string,
|
||||
): Promise<void> {
|
||||
const affectedProfiles = await tx.accountGrant.findMany({
|
||||
where: {
|
||||
OR: [{ grantorId: userId }, { granteeId: userId }],
|
||||
grantor: { managedProfileAt: { not: null } },
|
||||
},
|
||||
select: { grantorId: true },
|
||||
distinct: ["grantorId"],
|
||||
orderBy: { grantorId: "asc" },
|
||||
});
|
||||
|
||||
// Acquire every key before reading the final state. Sorting is required
|
||||
// because a wipe can affect more than one managed profile.
|
||||
for (const { grantorId } of affectedProfiles) {
|
||||
await withManagedProfileLock(tx, grantorId, async () => undefined);
|
||||
}
|
||||
|
||||
// Re-read after all locks are held. A concurrent acceptance, revocation,
|
||||
// expiry reduction, or notification claim uses the same key and is now
|
||||
// ordered entirely before or after this transaction.
|
||||
const now = new Date();
|
||||
for (const { grantorId } of affectedProfiles) {
|
||||
await withManagedProfileLock(tx, grantorId, async (profile) => {
|
||||
if (!profile?.managedProfileAt) return;
|
||||
|
||||
const [activeGuardians, guardiansRemovedByWipe] = await Promise.all([
|
||||
tx.accountGrant.count({
|
||||
where: { grantorId: profile.id, ...activeGuardianWhere(now) },
|
||||
}),
|
||||
tx.accountGrant.count({
|
||||
where: {
|
||||
AND: [
|
||||
{ grantorId: profile.id },
|
||||
{ OR: [{ grantorId: userId }, { granteeId: userId }] },
|
||||
activeGuardianWhere(now),
|
||||
],
|
||||
},
|
||||
}),
|
||||
]);
|
||||
|
||||
if (activeGuardians <= guardiansRemovedByWipe) {
|
||||
throw new LastManagedGuardianError();
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export async function deleteManagedProfile(input: {
|
||||
profileId: string;
|
||||
guardianId: string;
|
||||
|
||||
@@ -44,6 +44,7 @@ vi.mock("@/lib/logging/context", () => ({
|
||||
const sendViaTelegramMock = vi.fn();
|
||||
const sendViaNtfyMock = vi.fn();
|
||||
const sendViaWebPushMock = vi.fn();
|
||||
const sendViaApnsMock = vi.fn();
|
||||
|
||||
vi.mock("@/lib/notifications/senders/telegram", () => ({
|
||||
sendViaTelegram: (...args: unknown[]) => sendViaTelegramMock(...args),
|
||||
@@ -54,6 +55,9 @@ vi.mock("@/lib/notifications/senders/ntfy", () => ({
|
||||
vi.mock("@/lib/notifications/senders/web-push", () => ({
|
||||
sendViaWebPush: (...args: unknown[]) => sendViaWebPushMock(...args),
|
||||
}));
|
||||
vi.mock("@/lib/notifications/senders/apns", () => ({
|
||||
sendViaApns: (...args: unknown[]) => sendViaApnsMock(...args),
|
||||
}));
|
||||
|
||||
import { dispatchNotification } from "@/lib/notifications/dispatcher";
|
||||
import { prisma } from "@/lib/db";
|
||||
@@ -62,7 +66,7 @@ import { auditLog } from "@/lib/auth/audit";
|
||||
type MockChannel = {
|
||||
id: string;
|
||||
userId: string;
|
||||
type: "TELEGRAM" | "NTFY" | "WEB_PUSH";
|
||||
type: "TELEGRAM" | "NTFY" | "WEB_PUSH" | "APNS";
|
||||
enabled: boolean;
|
||||
config: string;
|
||||
consecutiveFailures: number;
|
||||
@@ -203,6 +207,36 @@ describe("dispatchNotification — soft reject (web-push 429)", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("dispatchNotification — bounded APNs fallback", () => {
|
||||
it("continues to the next channel after an APNs timeout", async () => {
|
||||
vi.mocked(prisma.notificationChannel.findMany).mockResolvedValueOnce([
|
||||
makeChannel({ id: "apns", type: "APNS" }),
|
||||
makeChannel({ id: "ntfy", type: "NTFY" }),
|
||||
] as never);
|
||||
sendViaApnsMock.mockResolvedValueOnce({
|
||||
ok: false,
|
||||
hardReject: false,
|
||||
reason: "apns_timeout",
|
||||
});
|
||||
sendViaNtfyMock.mockResolvedValueOnce({ ok: true });
|
||||
|
||||
const result = await dispatchNotification({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: "u-1",
|
||||
title: "t",
|
||||
message: "m",
|
||||
});
|
||||
|
||||
expect(sendViaApnsMock).toHaveBeenCalledTimes(1);
|
||||
expect(sendViaNtfyMock).toHaveBeenCalledTimes(1);
|
||||
expect(result).toMatchObject({
|
||||
dispatched: true,
|
||||
channelsAttempted: 2,
|
||||
channelsSucceeded: 1,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("dispatchNotification — give-up after 5 transient failures", () => {
|
||||
it("auto-disables on the 5th in-a-row failure", async () => {
|
||||
const channel = makeChannel({ type: "NTFY", consecutiveFailures: 4 });
|
||||
|
||||
@@ -164,6 +164,30 @@ describe("web-push reminder push — stable tag + badge", () => {
|
||||
);
|
||||
expect(body.tag).toBe("REMINDER");
|
||||
});
|
||||
|
||||
it("uses an ordinary landing route without actions for a managed Guardian", async () => {
|
||||
findManyMock.mockResolvedValue([SUB]);
|
||||
vi.spyOn(global, "fetch").mockResolvedValue(
|
||||
new Response(null, { status: 201 }),
|
||||
);
|
||||
|
||||
await sendViaWebPush("guardian-1", {
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: "record-1",
|
||||
recordUserId: "record-1",
|
||||
recipientUserId: "guardian-1",
|
||||
title: "Time for your dose",
|
||||
message: "Ramipril 5mg",
|
||||
urgent: true,
|
||||
metadata: { url: "/medications/med-1" },
|
||||
});
|
||||
|
||||
const body = JSON.parse(
|
||||
generateRequestDetailsMock.mock.calls[0][1] as string,
|
||||
);
|
||||
expect(body.url).toBe("/");
|
||||
expect(body.actions).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe("clear-on-taken push", () => {
|
||||
|
||||
@@ -1,9 +1,15 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
|
||||
import { prisma } from "@/lib/db";
|
||||
import {
|
||||
activeGuardianWhere,
|
||||
withManagedProfileLock,
|
||||
} from "@/lib/managed-profiles/lifecycle";
|
||||
import type { NotificationPayload } from "@/lib/notifications/types";
|
||||
import type { Prisma } from "@/generated/prisma/client";
|
||||
import type {
|
||||
ChannelType,
|
||||
NotificationPayload,
|
||||
} from "@/lib/notifications/types";
|
||||
|
||||
export interface NotificationDeliveryIdentity {
|
||||
recordUserId: string;
|
||||
@@ -11,16 +17,79 @@ export interface NotificationDeliveryIdentity {
|
||||
managed: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* The policy is deliberately closed. `SYSTEM_ALERT` by itself is never a
|
||||
* Guardian event: the safety-floor producer must name its narrow admission
|
||||
* marker, while illness red flags continue to remain record-local.
|
||||
*/
|
||||
export function isManagedGuardianFanoutAllowed(
|
||||
payload: Pick<NotificationPayload, "eventType" | "managedFanoutEvent">,
|
||||
): boolean {
|
||||
if (payload.managedFanoutEvent === "SAFETY_FLOOR_ALERT") {
|
||||
return payload.eventType === "SYSTEM_ALERT";
|
||||
}
|
||||
|
||||
return (
|
||||
payload.managedFanoutEvent === undefined &&
|
||||
(payload.eventType === "MEDICATION_REMINDER" ||
|
||||
payload.eventType === "MEASUREMENT_REMINDER" ||
|
||||
payload.eventType === "MEDICATION_LOW_STOCK")
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Enumerate candidate Guardians only for a legacy record-addressed delivery.
|
||||
* Each candidate is still re-authorized under the managed-profile advisory
|
||||
* lock immediately before its own channel selection and provider egress.
|
||||
*/
|
||||
export async function resolveManagedGuardianRecipientIds(
|
||||
payload: NotificationPayload,
|
||||
): Promise<string[] | null> {
|
||||
if (
|
||||
payload.recordUserId !== undefined ||
|
||||
payload.recipientUserId !== undefined ||
|
||||
!isManagedGuardianFanoutAllowed(payload)
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const record = await prisma.user.findUnique({
|
||||
where: { id: payload.userId },
|
||||
select: { managedProfileAt: true },
|
||||
});
|
||||
if (record?.managedProfileAt == null) return null;
|
||||
|
||||
const grants = await prisma.accountGrant.findMany({
|
||||
where: {
|
||||
grantorId: payload.userId,
|
||||
...activeGuardianWhere(new Date()),
|
||||
grantee: { managedProfileAt: null },
|
||||
},
|
||||
select: {
|
||||
granteeId: true,
|
||||
grantee: { select: { managedProfileAt: true } },
|
||||
},
|
||||
});
|
||||
|
||||
return grants
|
||||
.filter(
|
||||
(grant) =>
|
||||
grant.granteeId !== payload.userId &&
|
||||
grant.grantee.managedProfileAt == null,
|
||||
)
|
||||
.map((grant) => grant.granteeId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve notification identities without treating a managed record as its
|
||||
* own destination. Legacy payloads remain valid only for ordinary self
|
||||
* delivery; a managed record must name an active Guardian explicitly.
|
||||
* delivery; a managed record must name an active Guardian explicitly. This is
|
||||
* a preflight for recipient channel selection; managed delivery is authorized
|
||||
* again under the same lock at each provider egress claim below.
|
||||
*
|
||||
* Cross-principal authorization linearizes at the locked grant read below,
|
||||
* immediately before the dispatcher selects a channel. A revocation that
|
||||
* commits before that read prevents delivery; one that commits afterwards
|
||||
* affects later dispatches but cannot recall provider work already in flight.
|
||||
* The provider call deliberately runs outside the short database transaction.
|
||||
* Cross-principal authorization preflight uses the locked grant read below.
|
||||
* The final authorization is a separate short transaction that commits a
|
||||
* durable egress claim immediately before the provider operation starts.
|
||||
*/
|
||||
export async function resolveNotificationDeliveryIdentity(
|
||||
payload: NotificationPayload,
|
||||
@@ -50,7 +119,100 @@ export async function resolveNotificationDeliveryIdentity(
|
||||
|
||||
if (recipientUserId === recordUserId) return null;
|
||||
|
||||
return prisma.$transaction(async (tx) =>
|
||||
if (!isManagedGuardianFanoutAllowed(payload)) return null;
|
||||
|
||||
return withManagedGuardianAuthorization(recordUserId, recipientUserId, () =>
|
||||
Promise.resolve({ recordUserId, recipientUserId, managed: true }),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create the final authorization claim for one managed Guardian channel.
|
||||
*
|
||||
* The claim commits while holding the same lifecycle lock as Guardian
|
||||
* revocation and expiry. Its commit is the linearization point: if a
|
||||
* revocation commits first, this returns null and no provider operation may
|
||||
* start. If the claim commits first, it authorizes one provider operation
|
||||
* that is already in flight from the lifecycle's perspective. The provider
|
||||
* operation runs only after this transaction commits, so slow networks never
|
||||
* retain a database transaction, advisory lock, or pooled connection. This is
|
||||
* the logical send time, not a promise that a later revocation or expiry can
|
||||
* recall an already-authorized provider operation before its first byte.
|
||||
*/
|
||||
export async function claimManagedGuardianEgressAuthorization(
|
||||
delivery: NotificationDeliveryIdentity,
|
||||
channel: ChannelType,
|
||||
eventType: string,
|
||||
): Promise<{ id: string } | null> {
|
||||
if (!delivery.managed) return null;
|
||||
|
||||
return withManagedGuardianAuthorization(
|
||||
delivery.recordUserId,
|
||||
delivery.recipientUserId,
|
||||
async (tx) => {
|
||||
const id = randomUUID();
|
||||
const claims = await tx.$queryRaw<Array<{ id: string }>>`
|
||||
INSERT INTO "notification_egress_authorizations" (
|
||||
"id",
|
||||
"record_user_id",
|
||||
"recipient_user_id",
|
||||
"channel",
|
||||
"event_type",
|
||||
"authorized_at"
|
||||
)
|
||||
SELECT
|
||||
${id},
|
||||
${delivery.recordUserId},
|
||||
${delivery.recipientUserId},
|
||||
${channel},
|
||||
${eventType},
|
||||
clock_timestamp()
|
||||
WHERE EXISTS (
|
||||
SELECT 1
|
||||
FROM "account_grants"
|
||||
WHERE "grantor_id" = ${delivery.recordUserId}
|
||||
AND "grantee_id" = ${delivery.recipientUserId}
|
||||
AND "access" = 'MANAGE'::"account_grant_access"
|
||||
AND "accepted_at" IS NOT NULL
|
||||
AND "revoked_at" IS NULL
|
||||
AND (
|
||||
"expires_at" IS NULL
|
||||
OR "expires_at" > clock_timestamp()
|
||||
)
|
||||
)
|
||||
RETURNING "id"
|
||||
`;
|
||||
return claims[0] ?? null;
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Record a completed provider operation outside the lifecycle transaction.
|
||||
* This is observability only: authorization always rechecks the active grant,
|
||||
* and every retry creates a new claim regardless of a prior outcome.
|
||||
*/
|
||||
export async function completeManagedGuardianEgressAuthorization(
|
||||
authorizationId: string,
|
||||
outcome: "ok" | "hard_reject" | "transient_failure" | "sender_threw",
|
||||
): Promise<void> {
|
||||
try {
|
||||
await prisma.notificationEgressAuthorization.update({
|
||||
where: { id: authorizationId },
|
||||
data: { completedAt: new Date(), outcome },
|
||||
});
|
||||
} catch {
|
||||
// The claim is intentionally durable even if the best-effort completion
|
||||
// marker cannot be written after an otherwise independent provider call.
|
||||
}
|
||||
}
|
||||
|
||||
async function withManagedGuardianAuthorization<T>(
|
||||
recordUserId: string,
|
||||
recipientUserId: string,
|
||||
operation: (tx: Prisma.TransactionClient) => Promise<T>,
|
||||
): Promise<T | null> {
|
||||
return prisma.$transaction((tx) =>
|
||||
withManagedProfileLock(tx, recordUserId, async (lockedRecord) => {
|
||||
if (lockedRecord?.managedProfileAt == null) return null;
|
||||
|
||||
@@ -70,7 +232,7 @@ export async function resolveNotificationDeliveryIdentity(
|
||||
});
|
||||
if (!guardian) return null;
|
||||
|
||||
return { recordUserId, recipientUserId, managed: true };
|
||||
return operation(tx);
|
||||
}),
|
||||
);
|
||||
}
|
||||
@@ -41,7 +41,7 @@ import { getEvent } from "@/lib/logging/context";
|
||||
import { defaultLocale, type Locale } from "@/lib/i18n/config";
|
||||
import { getServerTranslator } from "@/lib/i18n/server-translator";
|
||||
import { dispatchNotification } from "@/lib/notifications/dispatcher";
|
||||
import type { EventType } from "@/lib/notifications/types";
|
||||
import type { EventType, NotificationPayload } from "@/lib/notifications/types";
|
||||
|
||||
interface DispatchLocalisedOptions {
|
||||
userId: string;
|
||||
@@ -61,6 +61,8 @@ interface DispatchLocalisedOptions {
|
||||
eventType?: EventType;
|
||||
/** Channel-specific extras forwarded to the dispatcher metadata. */
|
||||
metadata?: Record<string, unknown>;
|
||||
/** Closed internal marker for an admitted managed-record event. */
|
||||
managedFanoutEvent?: NotificationPayload["managedFanoutEvent"];
|
||||
/**
|
||||
* Escalate to every channel's highest urgency (v1.18.4). Forwarded to
|
||||
* the dispatcher so APNs goes time-sensitive, ntfy max, Web Push
|
||||
@@ -152,10 +154,14 @@ export async function dispatchLocalisedNotification(
|
||||
const locale = await resolveRecipientLocale(
|
||||
opts.recipientUserId ?? opts.userId,
|
||||
);
|
||||
const t = getServerTranslator(locale).t;
|
||||
|
||||
const title = t(opts.titleKey, opts.params);
|
||||
const message = t(opts.messageKey, opts.params);
|
||||
const renderForRecipient = (recipientLocale: Locale) => {
|
||||
const t = getServerTranslator(recipientLocale).t;
|
||||
return {
|
||||
title: t(opts.titleKey, opts.params),
|
||||
message: t(opts.messageKey, opts.params),
|
||||
};
|
||||
};
|
||||
const { title, message } = renderForRecipient(locale);
|
||||
|
||||
// Warn when a translation key falls back to its own raw string —
|
||||
// that signals a missing entry in `messages/{locale}.json` (or a
|
||||
@@ -181,7 +187,9 @@ export async function dispatchLocalisedNotification(
|
||||
recipientUserId: opts.recipientUserId,
|
||||
title,
|
||||
message,
|
||||
renderForRecipient,
|
||||
metadata,
|
||||
managedFanoutEvent: opts.managedFanoutEvent,
|
||||
urgent: opts.urgent,
|
||||
});
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
import { prisma } from "@/lib/db";
|
||||
import { decrypt, encrypt } from "@/lib/crypto";
|
||||
import { defaultLocale, locales, type Locale } from "@/lib/i18n/config";
|
||||
import type {
|
||||
NotificationPayload,
|
||||
TelegramChannelConfig,
|
||||
@@ -29,7 +30,12 @@ import {
|
||||
hasClientManagedApnsGate,
|
||||
} from "@/lib/notifications/client-managed-apns";
|
||||
import { recordPushAttempt } from "@/lib/notifications/senders/push-attempt-record";
|
||||
import { resolveNotificationDeliveryIdentity } from "@/lib/notifications/delivery-identity";
|
||||
import {
|
||||
claimManagedGuardianEgressAuthorization,
|
||||
completeManagedGuardianEgressAuthorization,
|
||||
resolveManagedGuardianRecipientIds,
|
||||
resolveNotificationDeliveryIdentity,
|
||||
} from "@/lib/notifications/delivery-identity";
|
||||
|
||||
/**
|
||||
* Dispatch a notification to all enabled channels for a user.
|
||||
@@ -57,6 +63,26 @@ export interface DispatchOutcome {
|
||||
channelsSucceeded: number;
|
||||
}
|
||||
|
||||
function isLocale(value: string | null | undefined): value is Locale {
|
||||
return (locales as readonly string[]).includes(value ?? "");
|
||||
}
|
||||
|
||||
/** Render record-owned content only after the delivery recipient is known. */
|
||||
async function renderForRecipient(
|
||||
payload: NotificationPayload,
|
||||
recipientUserId: string,
|
||||
): Promise<NotificationPayload> {
|
||||
const { renderForRecipient: render, ...deliveryPayload } = payload;
|
||||
if (!render) return deliveryPayload;
|
||||
|
||||
const recipient = await prisma.user.findUnique({
|
||||
where: { id: recipientUserId },
|
||||
select: { locale: true },
|
||||
});
|
||||
const locale = isLocale(recipient?.locale) ? recipient.locale : defaultLocale;
|
||||
return { ...deliveryPayload, ...render(locale) };
|
||||
}
|
||||
|
||||
export async function dispatchNotification(
|
||||
payload: NotificationPayload,
|
||||
): Promise<DispatchOutcome> {
|
||||
@@ -64,6 +90,30 @@ export async function dispatchNotification(
|
||||
let channelsSucceeded = 0;
|
||||
|
||||
try {
|
||||
const guardianRecipientIds =
|
||||
await resolveManagedGuardianRecipientIds(payload);
|
||||
if (guardianRecipientIds !== null) {
|
||||
const outcomes = await Promise.all(
|
||||
guardianRecipientIds.map((recipientUserId) =>
|
||||
dispatchNotification({
|
||||
...payload,
|
||||
recordUserId: payload.userId,
|
||||
recipientUserId,
|
||||
}),
|
||||
),
|
||||
);
|
||||
return outcomes.reduce<DispatchOutcome>(
|
||||
(aggregate, outcome) => ({
|
||||
dispatched: aggregate.dispatched || outcome.dispatched,
|
||||
channelsAttempted:
|
||||
aggregate.channelsAttempted + outcome.channelsAttempted,
|
||||
channelsSucceeded:
|
||||
aggregate.channelsSucceeded + outcome.channelsSucceeded,
|
||||
}),
|
||||
{ dispatched: false, channelsAttempted: 0, channelsSucceeded: 0 },
|
||||
);
|
||||
}
|
||||
|
||||
const delivery = await resolveNotificationDeliveryIdentity(payload);
|
||||
if (!delivery) {
|
||||
getEvent()?.addWarning("Notification delivery identity was invalid");
|
||||
@@ -74,11 +124,15 @@ export async function dispatchNotification(
|
||||
recordUserId: delivery.recordUserId,
|
||||
recipientUserId: delivery.recipientUserId,
|
||||
};
|
||||
const recipientPayload = await renderForRecipient(
|
||||
deliveryPayload,
|
||||
delivery.recipientUserId,
|
||||
);
|
||||
const channels = await prisma.notificationChannel.findMany({
|
||||
where: { userId: delivery.recipientUserId, enabled: true },
|
||||
include: {
|
||||
preferences: {
|
||||
where: { eventType: deliveryPayload.eventType },
|
||||
where: { eventType: recipientPayload.eventType },
|
||||
},
|
||||
},
|
||||
});
|
||||
@@ -128,7 +182,7 @@ export async function dispatchNotification(
|
||||
},
|
||||
include: {
|
||||
preferences: {
|
||||
where: { eventType: deliveryPayload.eventType },
|
||||
where: { eventType: recipientPayload.eventType },
|
||||
},
|
||||
},
|
||||
});
|
||||
@@ -159,7 +213,7 @@ export async function dispatchNotification(
|
||||
// from /settings/notifications once they've seen the badge a few
|
||||
// times and decided they want push.
|
||||
let defaultEnabled =
|
||||
EVENT_DEFAULT_ENABLED[deliveryPayload.eventType as EventType] ?? true;
|
||||
EVENT_DEFAULT_ENABLED[recipientPayload.eventType as EventType] ?? true;
|
||||
|
||||
// v1.7.0 — MOOD_REMINDER single source of truth = the visible card.
|
||||
// The event used to default OFF, layered on top of the per-user
|
||||
@@ -170,7 +224,7 @@ export async function dispatchNotification(
|
||||
// opt-out (a `NotificationPreference` row with `enabled = false`) is
|
||||
// still honoured below via `pref.enabled` — only the no-row default
|
||||
// is derived from the card.
|
||||
if (deliveryPayload.eventType === "MOOD_REMINDER") {
|
||||
if (recipientPayload.eventType === "MOOD_REMINDER") {
|
||||
const user = await prisma.user.findUnique({
|
||||
where: { id: delivery.recipientUserId },
|
||||
select: { moodReminderEnabled: true },
|
||||
@@ -184,7 +238,7 @@ export async function dispatchNotification(
|
||||
let clientManagedApns: boolean | null = null;
|
||||
const resolveClientManagedApns = async (): Promise<boolean> => {
|
||||
if (clientManagedApns !== null) return clientManagedApns;
|
||||
const gate = CLIENT_MANAGED_APNS_EVENTS[deliveryPayload.eventType];
|
||||
const gate = CLIENT_MANAGED_APNS_EVENTS[recipientPayload.eventType];
|
||||
if (!gate) {
|
||||
clientManagedApns = false;
|
||||
return false;
|
||||
@@ -224,38 +278,77 @@ export async function dispatchNotification(
|
||||
if (
|
||||
channel.type === "APNS" &&
|
||||
!delivery.managed &&
|
||||
hasClientManagedApnsGate(deliveryPayload.eventType) &&
|
||||
hasClientManagedApnsGate(recipientPayload.eventType) &&
|
||||
(await resolveClientManagedApns())
|
||||
) {
|
||||
const gate = CLIENT_MANAGED_APNS_EVENTS[deliveryPayload.eventType];
|
||||
getEvent()?.addMeta(gate.metaKey, gate.tag(deliveryPayload.metadata));
|
||||
const gate = CLIENT_MANAGED_APNS_EVENTS[recipientPayload.eventType];
|
||||
getEvent()?.addMeta(gate.metaKey, gate.tag(recipientPayload.metadata));
|
||||
if (gate.detail && gate.detailKey) {
|
||||
getEvent()?.addMeta(
|
||||
gate.detailKey,
|
||||
gate.detail(deliveryPayload.userId, deliveryPayload.metadata),
|
||||
gate.detail(recipientPayload.userId, recipientPayload.metadata),
|
||||
);
|
||||
}
|
||||
recordPushAttempt({
|
||||
recordUserId: delivery.recordUserId,
|
||||
recipientUserId: delivery.recipientUserId,
|
||||
channel: "APNS",
|
||||
eventType: deliveryPayload.eventType,
|
||||
eventType: recipientPayload.eventType,
|
||||
result: "skipped",
|
||||
reason: "client_managed",
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
channelsAttempted += 1;
|
||||
|
||||
try {
|
||||
const outcome = await sendToChannel(
|
||||
const prepared = prepareChannelSend(
|
||||
channel.type as ChannelType,
|
||||
channel.config,
|
||||
delivery.recipientUserId,
|
||||
deliveryPayload,
|
||||
recipientPayload,
|
||||
);
|
||||
let outcome: SendOutcome;
|
||||
if ("outcome" in prepared) {
|
||||
outcome = prepared.outcome;
|
||||
channelsAttempted += 1;
|
||||
} else {
|
||||
const authorization = delivery.managed
|
||||
? await claimManagedGuardianEgressAuthorization(
|
||||
delivery,
|
||||
channel.type as ChannelType,
|
||||
recipientPayload.eventType,
|
||||
)
|
||||
: null;
|
||||
if (delivery.managed && authorization === null) {
|
||||
getEvent()?.addWarning(
|
||||
"Guardian notification authorization changed before channel egress",
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
channelsAttempted += 1;
|
||||
try {
|
||||
outcome = await prepared.send();
|
||||
} catch (err) {
|
||||
if (authorization) {
|
||||
await completeManagedGuardianEgressAuthorization(
|
||||
authorization.id,
|
||||
"sender_threw",
|
||||
);
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
if (authorization) {
|
||||
await completeManagedGuardianEgressAuthorization(
|
||||
authorization.id,
|
||||
outcome.ok
|
||||
? "ok"
|
||||
: outcome.hardReject
|
||||
? "hard_reject"
|
||||
: "transient_failure",
|
||||
);
|
||||
}
|
||||
}
|
||||
if (outcome.ok) {
|
||||
await recordChannelSuccess({
|
||||
id: channel.id,
|
||||
@@ -324,12 +417,20 @@ export async function dispatchNotification(
|
||||
};
|
||||
}
|
||||
|
||||
async function sendToChannel(
|
||||
type PreparedChannelSend =
|
||||
{ send: () => Promise<SendOutcome> } | { outcome: SendOutcome };
|
||||
|
||||
/**
|
||||
* Decrypt and validate a recipient-owned channel before the final managed
|
||||
* Guardian authorization claim. The returned closure is the only code that
|
||||
* can begin provider I/O, and is invoked immediately after that claim commits.
|
||||
*/
|
||||
function prepareChannelSend(
|
||||
type: ChannelType,
|
||||
encryptedConfig: string,
|
||||
recipientUserId: string,
|
||||
payload: NotificationPayload,
|
||||
): Promise<SendOutcome> {
|
||||
): PreparedChannelSend {
|
||||
let decrypted: string;
|
||||
try {
|
||||
decrypted = decrypt(encryptedConfig);
|
||||
@@ -343,9 +444,11 @@ async function sendToChannel(
|
||||
// a key-rotation gap every channel would otherwise auto-disable for the
|
||||
// wrong reason.
|
||||
return {
|
||||
ok: false,
|
||||
hardReject: true,
|
||||
reason: `${type.toLowerCase()}_config_decrypt_failed`,
|
||||
outcome: {
|
||||
ok: false,
|
||||
hardReject: true,
|
||||
reason: `${type.toLowerCase()}_config_decrypt_failed`,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -358,12 +461,14 @@ async function sendToChannel(
|
||||
getEvent()?.addWarning("Failed to parse Telegram channel config");
|
||||
// Malformed config decodes the same on every retry — permanent.
|
||||
return {
|
||||
ok: false,
|
||||
hardReject: true,
|
||||
reason: "telegram_config_parse_failed",
|
||||
outcome: {
|
||||
ok: false,
|
||||
hardReject: true,
|
||||
reason: "telegram_config_parse_failed",
|
||||
},
|
||||
};
|
||||
}
|
||||
return sendViaTelegram(config, payload);
|
||||
return { send: () => sendViaTelegram(config, payload) };
|
||||
}
|
||||
case "NTFY": {
|
||||
let config: NtfyChannelConfig;
|
||||
@@ -373,12 +478,14 @@ async function sendToChannel(
|
||||
getEvent()?.addWarning("Failed to parse ntfy channel config");
|
||||
// Malformed config decodes the same on every retry — permanent.
|
||||
return {
|
||||
ok: false,
|
||||
hardReject: true,
|
||||
reason: "ntfy_config_parse_failed",
|
||||
outcome: {
|
||||
ok: false,
|
||||
hardReject: true,
|
||||
reason: "ntfy_config_parse_failed",
|
||||
},
|
||||
};
|
||||
}
|
||||
return sendViaNtfy(config, payload);
|
||||
return { send: () => sendViaNtfy(config, payload) };
|
||||
}
|
||||
case "WEBHOOK": {
|
||||
let config: WebhookChannelConfig;
|
||||
@@ -388,12 +495,14 @@ async function sendToChannel(
|
||||
getEvent()?.addWarning("Failed to parse webhook channel config");
|
||||
// Malformed config decodes the same on every retry — permanent.
|
||||
return {
|
||||
ok: false,
|
||||
hardReject: true,
|
||||
reason: "webhook_config_parse_failed",
|
||||
outcome: {
|
||||
ok: false,
|
||||
hardReject: true,
|
||||
reason: "webhook_config_parse_failed",
|
||||
},
|
||||
};
|
||||
}
|
||||
return sendViaWebhook(config, payload);
|
||||
return { send: () => sendViaWebhook(config, payload) };
|
||||
}
|
||||
case "EMAIL": {
|
||||
let config: EmailChannelConfig;
|
||||
@@ -403,15 +512,17 @@ async function sendToChannel(
|
||||
getEvent()?.addWarning("Failed to parse email channel config");
|
||||
// Malformed config decodes the same on every retry — permanent.
|
||||
return {
|
||||
ok: false,
|
||||
hardReject: true,
|
||||
reason: "email_config_parse_failed",
|
||||
outcome: {
|
||||
ok: false,
|
||||
hardReject: true,
|
||||
reason: "email_config_parse_failed",
|
||||
},
|
||||
};
|
||||
}
|
||||
return sendViaEmail(config, payload);
|
||||
return { send: () => sendViaEmail(config, payload) };
|
||||
}
|
||||
case "WEB_PUSH": {
|
||||
return sendViaWebPush(recipientUserId, payload);
|
||||
return { send: () => sendViaWebPush(recipientUserId, payload) };
|
||||
}
|
||||
case "APNS": {
|
||||
// APNs config lives on the Device row, not in the channel config.
|
||||
@@ -419,14 +530,16 @@ async function sendToChannel(
|
||||
// we just keep the symmetric path so future per-user overrides
|
||||
// (e.g. a custom APNs topic) can hang off the channel without
|
||||
// changing the dispatcher shape.
|
||||
return sendViaApns(recipientUserId, payload);
|
||||
return { send: () => sendViaApns(recipientUserId, payload) };
|
||||
}
|
||||
default:
|
||||
getEvent()?.addWarning(`Unknown notification channel type: ${type}`);
|
||||
return {
|
||||
ok: false,
|
||||
hardReject: false,
|
||||
reason: "unknown_channel_type",
|
||||
outcome: {
|
||||
ok: false,
|
||||
hardReject: false,
|
||||
reason: "unknown_channel_type",
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
import type { NotificationPayload } from "@/lib/notifications/types";
|
||||
|
||||
/**
|
||||
* A cross-account notification can only reach a Guardian after the
|
||||
* dispatcher has revalidated the marked record and its active MANAGE grant.
|
||||
* Senders use this narrow invariant to remove record-mutating affordances.
|
||||
*/
|
||||
export function isManagedGuardianDelivery(
|
||||
payload: Pick<NotificationPayload, "recordUserId" | "recipientUserId">,
|
||||
): boolean {
|
||||
return (
|
||||
payload.recordUserId !== undefined &&
|
||||
payload.recipientUserId !== undefined &&
|
||||
payload.recordUserId !== payload.recipientUserId
|
||||
);
|
||||
}
|
||||
@@ -128,6 +128,7 @@ beforeEach(() => {
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
setEnv();
|
||||
resetApnsForTesting();
|
||||
});
|
||||
@@ -367,6 +368,25 @@ describe("sendApnsPush — failure classification", () => {
|
||||
expect(r.reason).toBe("apns_network_error");
|
||||
expect(r.shouldDisable).toBeFalsy();
|
||||
});
|
||||
|
||||
it("bounds a provider send that never settles", async () => {
|
||||
vi.useFakeTimers();
|
||||
sendMock.mockImplementationOnce(() => new Promise(() => undefined));
|
||||
|
||||
const result = sendApnsPush({
|
||||
deviceToken: "abc",
|
||||
environment: "sandbox",
|
||||
payload: { alert: { title: "t", body: "b" } },
|
||||
});
|
||||
|
||||
await vi.advanceTimersByTimeAsync(5_000);
|
||||
|
||||
await expect(result).resolves.toMatchObject({
|
||||
ok: false,
|
||||
reason: "apns_timeout",
|
||||
shouldDisable: false,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("sendApnsPush — config disabled paths", () => {
|
||||
@@ -610,6 +630,31 @@ describe("sendViaApns — dispatcher fan-out", () => {
|
||||
expect(note.payload.eventType).toBe("MEDICATION_REMINDER");
|
||||
});
|
||||
|
||||
it("omits action metadata and categories for a managed Guardian", async () => {
|
||||
vi.mocked(prisma.device.findMany).mockResolvedValueOnce([
|
||||
{ id: "d1", apnsToken: "tok-a", apnsEnvironment: "sandbox" },
|
||||
] as never);
|
||||
sendMock.mockResolvedValueOnce({ sent: [{ device: "tok-a" }], failed: [] });
|
||||
|
||||
await sendViaApns("guardian-1", {
|
||||
title: "t",
|
||||
message: "m",
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
recordUserId: "record-1",
|
||||
recipientUserId: "guardian-1",
|
||||
metadata: {
|
||||
medicationId: "med-1",
|
||||
scheduleId: "sched-1",
|
||||
reminderId: "rem-1",
|
||||
url: "/medications/med-1",
|
||||
},
|
||||
});
|
||||
|
||||
const note = sendMock.mock.calls[0][0];
|
||||
expect(note.category).toBeUndefined();
|
||||
expect(note.payload).toEqual({ eventType: "MEDICATION_REMINDER" });
|
||||
});
|
||||
|
||||
it("forwards reminderId on MEASUREMENT_REMINDER and strips unlisted keys", async () => {
|
||||
// The iOS "Erledigt" action POSTs the reminder completion directly
|
||||
// from a foreground / handled notification, so the reminder's id must
|
||||
|
||||
@@ -200,6 +200,26 @@ describe("sendViaTelegram — per-slot delete scope (H2)", () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it("omits a record's inline actions for a managed Guardian", async () => {
|
||||
await sendViaTelegram(
|
||||
{ botToken: "bot-token", chatId: "guardian-chat" },
|
||||
{
|
||||
...reminderPayload({
|
||||
replyMarkup: {
|
||||
inline_keyboard: [[{ text: "Take", callback_data: "taken:med-1" }]],
|
||||
},
|
||||
}),
|
||||
userId: "managed-record",
|
||||
recordUserId: "managed-record",
|
||||
recipientUserId: "guardian-1",
|
||||
},
|
||||
);
|
||||
|
||||
expect(
|
||||
sendTelegramMessageMock.mock.calls[0][3].replyMarkup,
|
||||
).toBeUndefined();
|
||||
});
|
||||
|
||||
it("updates one recipient slot when its Telegram chat changes", async () => {
|
||||
findUniqueMock.mockResolvedValue({ chatId: "old-chat", messageId: 123 });
|
||||
|
||||
|
||||
@@ -146,6 +146,24 @@ describe("ntfy urgent mapping", () => {
|
||||
expect(headers.Priority).toBe("5");
|
||||
expect(headers.Tags).toBe("reminder");
|
||||
});
|
||||
|
||||
it("does not send ntfy action headers to a managed Guardian", async () => {
|
||||
safeFetchMock.mockResolvedValue({ ok: true, status: 200 });
|
||||
await sendViaNtfy(
|
||||
{ serverUrl: "https://ntfy.example.com", topic: "t" },
|
||||
payload({
|
||||
userId: "record-1",
|
||||
recordUserId: "record-1",
|
||||
recipientUserId: "guardian-1",
|
||||
}),
|
||||
);
|
||||
const headers = (
|
||||
safeFetchMock.mock.calls[0][1] as {
|
||||
headers: Record<string, string>;
|
||||
}
|
||||
).headers;
|
||||
expect(headers.Actions).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe("webhook urgent mapping", () => {
|
||||
|
||||
@@ -31,6 +31,7 @@ import { getEvent } from "@/lib/logging/context";
|
||||
import type { SendOutcome } from "@/lib/notifications/retry-policy";
|
||||
import { recordPushAttemptForPayload } from "@/lib/notifications/senders/push-attempt-record";
|
||||
import { plainPushText } from "@/lib/notifications/strip-emoji";
|
||||
import { isManagedGuardianDelivery } from "@/lib/notifications/managed-delivery";
|
||||
|
||||
export interface ApnsPayload {
|
||||
/** APNs `aps.alert.title` + `aps.alert.body`. */
|
||||
@@ -145,6 +146,53 @@ const PERMANENT_APNS_REASONS = new Set<string>([
|
||||
*/
|
||||
const APNS_COLLAPSE_ID_MAX_BYTES = 64;
|
||||
|
||||
/**
|
||||
* node-apn owns an HTTP/2 connection pool but exposes no AbortSignal on
|
||||
* Provider.send(). Bound the promise at this boundary so a stuck Apple
|
||||
* connection cannot prevent the dispatcher's later fallback channels.
|
||||
*/
|
||||
const APNS_PROVIDER_SEND_TIMEOUT_MS = 5_000;
|
||||
|
||||
class ApnsProviderTimeoutError extends Error {
|
||||
constructor() {
|
||||
super(
|
||||
`APNs provider send timed out after ${APNS_PROVIDER_SEND_TIMEOUT_MS}ms`,
|
||||
);
|
||||
this.name = "ApnsProviderTimeoutError";
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Observe both branches of the provider promise even after the timeout wins.
|
||||
* That prevents a late rejection from becoming an unhandled rejection while
|
||||
* retaining a hard upper bound for the dispatcher.
|
||||
*/
|
||||
function awaitBoundedApnsProviderSend<T>(operation: Promise<T>): Promise<T> {
|
||||
return new Promise((resolve, reject) => {
|
||||
let settled = false;
|
||||
const timeout = setTimeout(() => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
reject(new ApnsProviderTimeoutError());
|
||||
}, APNS_PROVIDER_SEND_TIMEOUT_MS);
|
||||
|
||||
operation.then(
|
||||
(result) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timeout);
|
||||
resolve(result);
|
||||
},
|
||||
(error: unknown) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timeout);
|
||||
reject(error);
|
||||
},
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
let cachedConfig: ApnsConfig | null | undefined;
|
||||
const providers = new Map<"sandbox" | "production", apn.Provider>();
|
||||
|
||||
@@ -466,16 +514,28 @@ export async function sendApnsPush(
|
||||
const start = performance.now();
|
||||
let result: Awaited<ReturnType<apn.Provider["send"]>>;
|
||||
try {
|
||||
result = await provider.send(note, input.deviceToken);
|
||||
result = await awaitBoundedApnsProviderSend(
|
||||
provider.send(note, input.deviceToken),
|
||||
);
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : "send_failed";
|
||||
const timedOut = err instanceof ApnsProviderTimeoutError;
|
||||
const message = timedOut
|
||||
? `APNs provider send timed out after ${APNS_PROVIDER_SEND_TIMEOUT_MS}ms`
|
||||
: err instanceof Error
|
||||
? err.message
|
||||
: "send_failed";
|
||||
getEvent()?.addExternalCall({
|
||||
service: "apns",
|
||||
method: "send",
|
||||
duration_ms: Math.round(performance.now() - start),
|
||||
error: message,
|
||||
error: timedOut ? "timeout" : message,
|
||||
});
|
||||
return { ok: false, reason: "apns_network_error", message };
|
||||
return {
|
||||
ok: false,
|
||||
reason: timedOut ? "apns_timeout" : "apns_network_error",
|
||||
shouldDisable: false,
|
||||
message,
|
||||
};
|
||||
}
|
||||
|
||||
const duration = Math.round(performance.now() - start);
|
||||
@@ -628,10 +688,7 @@ export async function sendViaApns(
|
||||
process.env.APNS_CRITICAL_ENTITLEMENT === "true";
|
||||
|
||||
let result: Awaited<ReturnType<typeof sendApnsPush>> | null = null;
|
||||
const managedDelivery =
|
||||
payload.recordUserId !== undefined &&
|
||||
payload.recipientUserId !== undefined &&
|
||||
payload.recordUserId !== payload.recipientUserId;
|
||||
const managedDelivery = isManagedGuardianDelivery(payload);
|
||||
// Discreet mode (cycle privacy): the lock-screen-visible routing
|
||||
// metadata must not carry the event name. Collapse the category /
|
||||
// threadId / collapseId and the data eventType to a generic value so
|
||||
@@ -671,10 +728,16 @@ export async function sendViaApns(
|
||||
// In discreet mode also drop cycle-semantic metadata (e.g. `phase`)
|
||||
// so a future cycle push can't name a cycle concept on the wire even
|
||||
// though the allowlist permits `phase` (QA LOW).
|
||||
// A Guardian may read a managed record but must never receive
|
||||
// an iOS action identifier or deep link that can mutate it.
|
||||
// Keep the ordinary event envelope so opening the notification
|
||||
// follows the normal authorised record-selection flow.
|
||||
...pickIosMetadata(
|
||||
payload.discreet
|
||||
? stripCycleMetadata(payload.metadata)
|
||||
: payload.metadata,
|
||||
managedDelivery
|
||||
? undefined
|
||||
: payload.discreet
|
||||
? stripCycleMetadata(payload.metadata)
|
||||
: payload.metadata,
|
||||
),
|
||||
},
|
||||
threadId: routingEvent,
|
||||
@@ -860,16 +923,28 @@ export async function sendApnsRawPush(
|
||||
const start = performance.now();
|
||||
let result: Awaited<ReturnType<apn.Provider["send"]>>;
|
||||
try {
|
||||
result = await provider.send(note, input.deviceToken);
|
||||
result = await awaitBoundedApnsProviderSend(
|
||||
provider.send(note, input.deviceToken),
|
||||
);
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : "send_failed";
|
||||
const timedOut = err instanceof ApnsProviderTimeoutError;
|
||||
const message = timedOut
|
||||
? `APNs provider send timed out after ${APNS_PROVIDER_SEND_TIMEOUT_MS}ms`
|
||||
: err instanceof Error
|
||||
? err.message
|
||||
: "send_failed";
|
||||
getEvent()?.addExternalCall({
|
||||
service: "apns",
|
||||
method: `send_${input.pushType}`,
|
||||
duration_ms: Math.round(performance.now() - start),
|
||||
error: message,
|
||||
error: timedOut ? "timeout" : message,
|
||||
});
|
||||
return { ok: false, reason: "apns_network_error", message };
|
||||
return {
|
||||
ok: false,
|
||||
reason: timedOut ? "apns_timeout" : "apns_network_error",
|
||||
shouldDisable: false,
|
||||
message,
|
||||
};
|
||||
}
|
||||
|
||||
const duration = Math.round(performance.now() - start);
|
||||
|
||||
@@ -9,6 +9,7 @@ import { recordPushAttemptForPayload } from "@/lib/notifications/senders/push-at
|
||||
import { isPublicUrl } from "@/lib/validations/notifications";
|
||||
import { safeFetch } from "@/lib/safe-fetch";
|
||||
import { plainPushText } from "@/lib/notifications/strip-emoji";
|
||||
import { isManagedGuardianDelivery } from "@/lib/notifications/managed-delivery";
|
||||
|
||||
/**
|
||||
* Send Web Push notification to all subscribed devices of a user.
|
||||
@@ -30,6 +31,7 @@ export async function sendViaWebPush(
|
||||
payload: NotificationPayload,
|
||||
): Promise<SendOutcome> {
|
||||
try {
|
||||
const managedDelivery = isManagedGuardianDelivery(payload);
|
||||
// Lazy import to avoid issues when web-push is not installed
|
||||
const webpush = await import("web-push");
|
||||
|
||||
@@ -116,6 +118,7 @@ export async function sendViaWebPush(
|
||||
// target. Only relative paths are honoured so a poisoned metadata
|
||||
// value can never point the SW at a foreign origin.
|
||||
url:
|
||||
!managedDelivery &&
|
||||
typeof payload.metadata?.url === "string" &&
|
||||
payload.metadata.url.startsWith("/")
|
||||
? payload.metadata.url
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import type { Locale } from "@/lib/i18n/config";
|
||||
|
||||
/**
|
||||
* Notification system type definitions.
|
||||
* Channels: Telegram, ntfy, Web Push
|
||||
@@ -125,6 +127,20 @@ export const EVENT_TYPES = [
|
||||
] as const;
|
||||
export type EventType = (typeof EVENT_TYPES)[number];
|
||||
|
||||
/**
|
||||
* The closed set of managed-record events that may reach Guardians. A
|
||||
* `SYSTEM_ALERT` needs the explicit safety-floor discriminator because
|
||||
* illness red flags use the same broad delivery event but do not fan out.
|
||||
*/
|
||||
export const MANAGED_GUARDIAN_FANOUT_EVENTS = [
|
||||
"MEDICATION_REMINDER",
|
||||
"MEASUREMENT_REMINDER",
|
||||
"MEDICATION_LOW_STOCK",
|
||||
"SAFETY_FLOOR_ALERT",
|
||||
] as const;
|
||||
export type ManagedGuardianFanoutEvent =
|
||||
(typeof MANAGED_GUARDIAN_FANOUT_EVENTS)[number];
|
||||
|
||||
/**
|
||||
* Per-event default policy when no `NotificationPreference` row
|
||||
* exists. The dispatcher reads this to decide whether to fire on a
|
||||
@@ -265,6 +281,21 @@ export interface NotificationPayload {
|
||||
message: string;
|
||||
/** Channel-specific extras (e.g. medicationId for Telegram inline buttons) */
|
||||
metadata?: Record<string, unknown>;
|
||||
/**
|
||||
* Internal admission marker for managed-record delivery. It is deliberately
|
||||
* absent from public request contracts: only the four allowlisted producer
|
||||
* paths may set it, and safety-floor is the only SYSTEM_ALERT admission.
|
||||
*/
|
||||
managedFanoutEvent?: ManagedGuardianFanoutEvent;
|
||||
/**
|
||||
* Internal rendering hook for record-owned notification content. The
|
||||
* dispatcher invokes it only after choosing a recipient, then removes it
|
||||
* before handing the payload to a channel adapter.
|
||||
*/
|
||||
renderForRecipient?: (locale: Locale) => {
|
||||
title: string;
|
||||
message: string;
|
||||
};
|
||||
/**
|
||||
* Discreet mode (cycle privacy). When true, the lock-screen-visible
|
||||
* routing metadata the senders would otherwise derive from `eventType`
|
||||
|
||||
@@ -107,6 +107,30 @@ async function cleanupReminderTracking(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A managed-record reminder is tracked under the Guardian recipient so a
|
||||
* replacement can delete the prior message. Its callback must never mutate
|
||||
* the managed record, even if a crafted Telegram update supplies action data.
|
||||
*/
|
||||
async function isManagedReminderCallback(input: {
|
||||
recipientUserId: string;
|
||||
chatId: string;
|
||||
messageId: number;
|
||||
}): Promise<boolean> {
|
||||
const reminder = await prisma.telegramReminderMessage.findFirst({
|
||||
where: {
|
||||
recipientUserId: input.recipientUserId,
|
||||
chatId: input.chatId,
|
||||
messageId: input.messageId,
|
||||
},
|
||||
select: { medication: { select: { userId: true } } },
|
||||
});
|
||||
return (
|
||||
reminder?.medication?.userId !== undefined &&
|
||||
reminder.medication.userId !== input.recipientUserId
|
||||
);
|
||||
}
|
||||
|
||||
// v1.19.0 — the auto-delete window dropped from 1 h to ~30 min and moved to
|
||||
// the shared `scheduleTelegramAutoDelete` helper so the inbound (free-text)
|
||||
// and outbound (unanswered prompt) paths agree on the window.
|
||||
@@ -379,6 +403,23 @@ export async function handleCallback(update: TelegramUpdate) {
|
||||
const data = callback.data ?? "";
|
||||
const messageId = callback.message?.message_id;
|
||||
|
||||
if (
|
||||
messageId !== undefined &&
|
||||
/^(taken|snooze|skip|ack):/.test(data) &&
|
||||
(await isManagedReminderCallback({
|
||||
recipientUserId: user.id,
|
||||
chatId,
|
||||
messageId,
|
||||
}))
|
||||
) {
|
||||
await answerTelegramCallbackQuery(
|
||||
botToken,
|
||||
callback.id,
|
||||
t("telegram.errorInvalidAction"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
if (data.startsWith("taken:")) {
|
||||
const medicationId = data.slice("taken:".length).trim();
|
||||
if (!medicationId) {
|
||||
|
||||
@@ -235,6 +235,42 @@ describe("APNs dispatcher integration", () => {
|
||||
expect(audit).toBeTruthy();
|
||||
});
|
||||
|
||||
it("bounds a hung APNs provider and falls through to Telegram", async () => {
|
||||
const { encrypt } = await import("@/lib/crypto");
|
||||
await seedApnsChannel();
|
||||
await seedDevice({ apnsToken: "hung-apns-token" });
|
||||
await getPrismaClient().notificationChannel.create({
|
||||
data: {
|
||||
userId: TEST_USER_ID,
|
||||
type: "TELEGRAM",
|
||||
enabled: true,
|
||||
config: encrypt(
|
||||
JSON.stringify({ botToken: "telegram-token", chatId: "chat-id" }),
|
||||
),
|
||||
},
|
||||
});
|
||||
|
||||
apnsSendMock.mockImplementationOnce(() => new Promise(() => undefined));
|
||||
telegramSendMock.mockResolvedValueOnce({ ok: true, messageId: 42 });
|
||||
|
||||
const { dispatchNotification } =
|
||||
await import("@/lib/notifications/dispatcher");
|
||||
const result = await dispatchNotification({
|
||||
userId: TEST_USER_ID,
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
title: "fallback",
|
||||
message: "fallback",
|
||||
});
|
||||
|
||||
expect(apnsSendMock).toHaveBeenCalledTimes(1);
|
||||
expect(telegramSendMock).toHaveBeenCalledTimes(1);
|
||||
expect(result).toMatchObject({
|
||||
dispatched: true,
|
||||
channelsAttempted: 2,
|
||||
channelsSucceeded: 1,
|
||||
});
|
||||
}, 7_000);
|
||||
|
||||
it("falls through to Telegram when APNS is disabled", async () => {
|
||||
const { encrypt } = await import("@/lib/crypto");
|
||||
// Pre-existing disabled APNS channel from a previous failure run.
|
||||
|
||||
@@ -0,0 +1,647 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const sent = vi.hoisted(() => ({
|
||||
deliveries: [] as Array<{
|
||||
channel: "NTFY" | "APNS" | "WEB_PUSH" | "TELEGRAM";
|
||||
recordUserId: string;
|
||||
recipientUserId: string;
|
||||
}>,
|
||||
ntfy: [] as Array<{
|
||||
recipientUserId: string;
|
||||
title: string;
|
||||
message: string;
|
||||
}>,
|
||||
apns: [] as Array<{ recipientUserId: string; recordUserId?: string }>,
|
||||
}));
|
||||
|
||||
const ntfyEgress = vi.hoisted(() => ({
|
||||
block: null as Promise<void> | null,
|
||||
onStart: null as (() => void) | null,
|
||||
error: null as Error | null,
|
||||
outcome: null as { ok: false; reason: string } | null,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/notifications/senders/ntfy", async () => {
|
||||
const { recordPushAttemptForPayload } = await vi.importActual<
|
||||
typeof import("@/lib/notifications/senders/push-attempt-record")
|
||||
>("@/lib/notifications/senders/push-attempt-record");
|
||||
return {
|
||||
sendViaNtfy: async (
|
||||
_config: unknown,
|
||||
payload: {
|
||||
userId: string;
|
||||
recordUserId?: string;
|
||||
recipientUserId?: string;
|
||||
eventType: string;
|
||||
title: string;
|
||||
message: string;
|
||||
},
|
||||
) => {
|
||||
ntfyEgress.onStart?.();
|
||||
if (ntfyEgress.block) await ntfyEgress.block;
|
||||
if (ntfyEgress.error) throw ntfyEgress.error;
|
||||
if (ntfyEgress.outcome) return ntfyEgress.outcome;
|
||||
const recipientUserId = payload.recipientUserId ?? payload.userId;
|
||||
sent.ntfy.push({
|
||||
recipientUserId,
|
||||
title: payload.title,
|
||||
message: payload.message,
|
||||
});
|
||||
sent.deliveries.push({
|
||||
channel: "NTFY",
|
||||
recordUserId: payload.recordUserId ?? payload.userId,
|
||||
recipientUserId,
|
||||
});
|
||||
recordPushAttemptForPayload(payload, recipientUserId, {
|
||||
userId: payload.userId,
|
||||
channel: "NTFY",
|
||||
eventType: payload.eventType,
|
||||
result: "ok",
|
||||
});
|
||||
return { ok: true };
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("@/lib/notifications/senders/apns", async () => {
|
||||
const { recordPushAttemptForPayload } = await vi.importActual<
|
||||
typeof import("@/lib/notifications/senders/push-attempt-record")
|
||||
>("@/lib/notifications/senders/push-attempt-record");
|
||||
return {
|
||||
sendViaApns: async (
|
||||
recipientUserId: string,
|
||||
payload: { userId: string; recordUserId?: string; eventType: string },
|
||||
) => {
|
||||
sent.apns.push({ recipientUserId, recordUserId: payload.recordUserId });
|
||||
sent.deliveries.push({
|
||||
channel: "APNS",
|
||||
recordUserId: payload.recordUserId ?? payload.userId,
|
||||
recipientUserId,
|
||||
});
|
||||
recordPushAttemptForPayload(payload, recipientUserId, {
|
||||
userId: payload.userId,
|
||||
channel: "APNS",
|
||||
eventType: payload.eventType,
|
||||
result: "ok",
|
||||
});
|
||||
return { ok: true };
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("@/lib/notifications/senders/web-push", async () => {
|
||||
const { recordPushAttemptForPayload } = await vi.importActual<
|
||||
typeof import("@/lib/notifications/senders/push-attempt-record")
|
||||
>("@/lib/notifications/senders/push-attempt-record");
|
||||
return {
|
||||
sendViaWebPush: async (
|
||||
recipientUserId: string,
|
||||
payload: { userId: string; recordUserId?: string; eventType: string },
|
||||
) => {
|
||||
sent.deliveries.push({
|
||||
channel: "WEB_PUSH",
|
||||
recordUserId: payload.recordUserId ?? payload.userId,
|
||||
recipientUserId,
|
||||
});
|
||||
recordPushAttemptForPayload(payload, recipientUserId, {
|
||||
userId: payload.userId,
|
||||
channel: "WEB_PUSH",
|
||||
eventType: payload.eventType,
|
||||
result: "ok",
|
||||
});
|
||||
return { ok: true };
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("@/lib/notifications/senders/telegram", async () => {
|
||||
const { recordPushAttemptForPayload } = await vi.importActual<
|
||||
typeof import("@/lib/notifications/senders/push-attempt-record")
|
||||
>("@/lib/notifications/senders/push-attempt-record");
|
||||
return {
|
||||
sendViaTelegram: async (
|
||||
_config: unknown,
|
||||
payload: {
|
||||
userId: string;
|
||||
recordUserId?: string;
|
||||
recipientUserId?: string;
|
||||
eventType: string;
|
||||
},
|
||||
) => {
|
||||
const recipientUserId = payload.recipientUserId ?? payload.userId;
|
||||
sent.deliveries.push({
|
||||
channel: "TELEGRAM",
|
||||
recordUserId: payload.recordUserId ?? payload.userId,
|
||||
recipientUserId,
|
||||
});
|
||||
recordPushAttemptForPayload(payload, recipientUserId, {
|
||||
userId: payload.userId,
|
||||
channel: "TELEGRAM",
|
||||
eventType: payload.eventType,
|
||||
result: "ok",
|
||||
});
|
||||
return { ok: true };
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
import { encrypt } from "@/lib/crypto";
|
||||
import type { Locale } from "@/lib/i18n/config";
|
||||
import { withManagedProfileLock } from "@/lib/managed-profiles/lifecycle";
|
||||
import { dispatchNotification } from "@/lib/notifications/dispatcher";
|
||||
import { revokeGrant } from "@/lib/sharing/grants";
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
|
||||
let sequence = 0;
|
||||
|
||||
async function createUser(input: {
|
||||
label: string;
|
||||
locale: "en" | "de";
|
||||
managed?: boolean;
|
||||
clientManaged?: boolean;
|
||||
}) {
|
||||
const suffix = sequence++;
|
||||
return getPrismaClient().user.create({
|
||||
data: {
|
||||
username: `${input.label}-${suffix}`,
|
||||
email: `${input.label}-${suffix}@example.test`,
|
||||
locale: input.locale,
|
||||
managedProfileAt: input.managed ? new Date() : null,
|
||||
notificationPrefs: input.clientManaged
|
||||
? { medication: { clientManaged: true } }
|
||||
: undefined,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async function addGuardian(recordUserId: string, recipientUserId: string) {
|
||||
await getPrismaClient().accountGrant.create({
|
||||
data: {
|
||||
grantorId: recordUserId,
|
||||
granteeId: recipientUserId,
|
||||
access: "MANAGE",
|
||||
acceptedAt: new Date(),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async function addChannel(
|
||||
userId: string,
|
||||
type: "NTFY" | "APNS" | "WEB_PUSH" | "TELEGRAM",
|
||||
enabled = true,
|
||||
) {
|
||||
await getPrismaClient().notificationChannel.create({
|
||||
data: {
|
||||
userId,
|
||||
type,
|
||||
enabled,
|
||||
config: encrypt("{}"),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
ntfyEgress.block = null;
|
||||
ntfyEgress.onStart = null;
|
||||
ntfyEgress.error = null;
|
||||
ntfyEgress.outcome = null;
|
||||
sent.deliveries.length = 0;
|
||||
sent.ntfy.length = 0;
|
||||
sent.apns.length = 0;
|
||||
await truncateAllTables(getPrismaClient());
|
||||
});
|
||||
|
||||
describe("Guardian fan-out (real Postgres)", () => {
|
||||
it("isolates recipient locale and global preference across two Guardians and two records", async () => {
|
||||
const [recordA, recordB, guardianEn, guardianDe] = await Promise.all([
|
||||
createUser({ label: "record-a", locale: "en", managed: true }),
|
||||
createUser({ label: "record-b", locale: "de", managed: true }),
|
||||
createUser({ label: "guardian-en", locale: "en" }),
|
||||
createUser({ label: "guardian-de", locale: "de" }),
|
||||
]);
|
||||
await Promise.all([
|
||||
addGuardian(recordA.id, guardianEn.id),
|
||||
addGuardian(recordA.id, guardianDe.id),
|
||||
addGuardian(recordB.id, guardianEn.id),
|
||||
addGuardian(recordB.id, guardianDe.id),
|
||||
addChannel(guardianEn.id, "NTFY"),
|
||||
addChannel(guardianDe.id, "NTFY"),
|
||||
]);
|
||||
|
||||
const renderForRecipient = (locale: Locale) => ({
|
||||
title: `title-${locale}`,
|
||||
message: `message-${locale}`,
|
||||
});
|
||||
await dispatchNotification({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: recordA.id,
|
||||
title: "record-a-title",
|
||||
message: "record-a-message",
|
||||
renderForRecipient,
|
||||
});
|
||||
|
||||
expect(sent.ntfy).toEqual(
|
||||
expect.arrayContaining([
|
||||
{
|
||||
recipientUserId: guardianEn.id,
|
||||
title: "title-en",
|
||||
message: "message-en",
|
||||
},
|
||||
{
|
||||
recipientUserId: guardianDe.id,
|
||||
title: "title-de",
|
||||
message: "message-de",
|
||||
},
|
||||
]),
|
||||
);
|
||||
|
||||
await getPrismaClient().notificationPreference.create({
|
||||
data: {
|
||||
channel: {
|
||||
connect: { userId_type: { userId: guardianEn.id, type: "NTFY" } },
|
||||
},
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
enabled: false,
|
||||
},
|
||||
});
|
||||
sent.ntfy.length = 0;
|
||||
|
||||
await dispatchNotification({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: recordB.id,
|
||||
title: "record-b-title",
|
||||
message: "record-b-message",
|
||||
renderForRecipient,
|
||||
});
|
||||
|
||||
expect(sent.ntfy).toEqual([
|
||||
{
|
||||
recipientUserId: guardianDe.id,
|
||||
title: "title-de",
|
||||
message: "message-de",
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it("bypasses client-managed suppression for each Guardian but not self delivery", async () => {
|
||||
const [record, guardianEn, guardianDe] = await Promise.all([
|
||||
createUser({ label: "record", locale: "en", managed: true }),
|
||||
createUser({
|
||||
label: "guardian-en",
|
||||
locale: "en",
|
||||
clientManaged: true,
|
||||
}),
|
||||
createUser({
|
||||
label: "guardian-de",
|
||||
locale: "de",
|
||||
clientManaged: true,
|
||||
}),
|
||||
]);
|
||||
await Promise.all([
|
||||
addGuardian(record.id, guardianEn.id),
|
||||
addGuardian(record.id, guardianDe.id),
|
||||
addChannel(guardianEn.id, "APNS"),
|
||||
addChannel(guardianDe.id, "APNS"),
|
||||
]);
|
||||
|
||||
await dispatchNotification({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: record.id,
|
||||
title: "record-title",
|
||||
message: "record-message",
|
||||
});
|
||||
|
||||
expect(sent.apns).toEqual(
|
||||
expect.arrayContaining([
|
||||
{ recipientUserId: guardianEn.id, recordUserId: record.id },
|
||||
{ recipientUserId: guardianDe.id, recordUserId: record.id },
|
||||
]),
|
||||
);
|
||||
});
|
||||
|
||||
it("writes the complete two-record, two-Guardian, four-channel attribution matrix", async () => {
|
||||
const [recordA, recordB, guardianOne, guardianTwo, ordinaryRecord] =
|
||||
await Promise.all([
|
||||
createUser({ label: "record-a", locale: "en", managed: true }),
|
||||
createUser({ label: "record-b", locale: "de", managed: true }),
|
||||
createUser({ label: "guardian-one", locale: "en" }),
|
||||
createUser({ label: "guardian-two", locale: "de" }),
|
||||
createUser({ label: "ordinary-record", locale: "en" }),
|
||||
]);
|
||||
const guardianIds = [guardianOne.id, guardianTwo.id];
|
||||
const recordIds = [recordA.id, recordB.id];
|
||||
const channelTypes = ["NTFY", "APNS", "WEB_PUSH", "TELEGRAM"] as const;
|
||||
|
||||
await Promise.all([
|
||||
...recordIds.flatMap((recordUserId) =>
|
||||
guardianIds.map((recipientUserId) =>
|
||||
addGuardian(recordUserId, recipientUserId),
|
||||
),
|
||||
),
|
||||
// An ordinary adult's MANAGE grant must never cause fan-out.
|
||||
addGuardian(ordinaryRecord.id, guardianOne.id),
|
||||
...guardianIds.flatMap((userId) =>
|
||||
channelTypes.map((type) => addChannel(userId, type)),
|
||||
),
|
||||
]);
|
||||
|
||||
await Promise.all(
|
||||
recordIds.map((userId) =>
|
||||
dispatchNotification({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId,
|
||||
title: "record reminder",
|
||||
message: "record message",
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
const expected = recordIds.flatMap((recordUserId) =>
|
||||
guardianIds.flatMap((recipientUserId) =>
|
||||
channelTypes.map((channel) => ({
|
||||
channel,
|
||||
recordUserId,
|
||||
recipientUserId,
|
||||
})),
|
||||
),
|
||||
);
|
||||
expect(sent.deliveries).toHaveLength(expected.length);
|
||||
expect(sent.deliveries).toEqual(expect.arrayContaining(expected));
|
||||
expect(
|
||||
sent.deliveries.some((delivery) =>
|
||||
recordIds.includes(delivery.recipientUserId),
|
||||
),
|
||||
).toBe(false);
|
||||
|
||||
await vi.waitFor(async () => {
|
||||
const attempts = await getPrismaClient().pushAttempt.findMany({
|
||||
where: { eventType: "MEDICATION_REMINDER" },
|
||||
select: { channel: true, recordUserId: true, recipientUserId: true },
|
||||
});
|
||||
expect(attempts).toHaveLength(expected.length);
|
||||
expect(attempts).toEqual(expect.arrayContaining(expected));
|
||||
});
|
||||
const claims =
|
||||
await getPrismaClient().notificationEgressAuthorization.findMany({
|
||||
where: { eventType: "MEDICATION_REMINDER" },
|
||||
select: {
|
||||
channel: true,
|
||||
recordUserId: true,
|
||||
recipientUserId: true,
|
||||
eventType: true,
|
||||
},
|
||||
});
|
||||
const expectedClaims = expected.map((claim) => ({
|
||||
...claim,
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
}));
|
||||
expect(claims).toHaveLength(expectedClaims.length);
|
||||
expect(claims).toEqual(expect.arrayContaining(expectedClaims));
|
||||
|
||||
await dispatchNotification({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: ordinaryRecord.id,
|
||||
title: "ordinary reminder",
|
||||
message: "ordinary message",
|
||||
});
|
||||
|
||||
expect(sent.deliveries).toHaveLength(expected.length);
|
||||
await expect(
|
||||
getPrismaClient().pushAttempt.count({
|
||||
where: { eventType: "MEDICATION_REMINDER" },
|
||||
}),
|
||||
).resolves.toBe(expected.length);
|
||||
});
|
||||
|
||||
it("allows Guardian revocation to commit while a claimed channel egress is in flight", async () => {
|
||||
const [record, recipient, reserveGuardian] = await Promise.all([
|
||||
createUser({ label: "record", locale: "en", managed: true }),
|
||||
createUser({ label: "recipient", locale: "en" }),
|
||||
createUser({ label: "reserve-guardian", locale: "de" }),
|
||||
]);
|
||||
await Promise.all([
|
||||
addGuardian(record.id, recipient.id),
|
||||
addGuardian(record.id, reserveGuardian.id),
|
||||
addChannel(recipient.id, "NTFY"),
|
||||
]);
|
||||
const recipientGrant =
|
||||
await getPrismaClient().accountGrant.findFirstOrThrow({
|
||||
where: { grantorId: record.id, granteeId: recipient.id },
|
||||
select: { id: true },
|
||||
});
|
||||
let releaseEgress!: () => void;
|
||||
let markEgressStarted!: () => void;
|
||||
ntfyEgress.block = new Promise<void>((resolve) => {
|
||||
releaseEgress = resolve;
|
||||
});
|
||||
const egressStarted = new Promise<void>((resolve) => {
|
||||
markEgressStarted = resolve;
|
||||
});
|
||||
ntfyEgress.onStart = markEgressStarted;
|
||||
|
||||
const delivery = dispatchNotification({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: record.id,
|
||||
title: "record reminder",
|
||||
message: "record message",
|
||||
});
|
||||
await egressStarted;
|
||||
await expect(
|
||||
getPrismaClient().notificationEgressAuthorization.count({
|
||||
where: {
|
||||
recordUserId: record.id,
|
||||
recipientUserId: recipient.id,
|
||||
channel: "NTFY",
|
||||
},
|
||||
}),
|
||||
).resolves.toBe(1);
|
||||
|
||||
const revocation = revokeGrant({
|
||||
grantId: recipientGrant.id,
|
||||
grantorId: record.id,
|
||||
});
|
||||
await expectPromiseToSettle(revocation);
|
||||
|
||||
releaseEgress();
|
||||
await delivery;
|
||||
});
|
||||
|
||||
it("allows Guardian expiry to commit while a claimed channel egress is in flight", async () => {
|
||||
const [record, recipient, reserveGuardian] = await Promise.all([
|
||||
createUser({ label: "record", locale: "en", managed: true }),
|
||||
createUser({ label: "recipient", locale: "en" }),
|
||||
createUser({ label: "reserve-guardian", locale: "de" }),
|
||||
]);
|
||||
await Promise.all([
|
||||
addGuardian(record.id, recipient.id),
|
||||
addGuardian(record.id, reserveGuardian.id),
|
||||
addChannel(recipient.id, "NTFY"),
|
||||
]);
|
||||
const recipientGrant =
|
||||
await getPrismaClient().accountGrant.findFirstOrThrow({
|
||||
where: { grantorId: record.id, granteeId: recipient.id },
|
||||
select: { id: true },
|
||||
});
|
||||
let releaseEgress!: () => void;
|
||||
let markEgressStarted!: () => void;
|
||||
ntfyEgress.block = new Promise<void>((resolve) => {
|
||||
releaseEgress = resolve;
|
||||
});
|
||||
const egressStarted = new Promise<void>((resolve) => {
|
||||
markEgressStarted = resolve;
|
||||
});
|
||||
ntfyEgress.onStart = markEgressStarted;
|
||||
|
||||
const delivery = dispatchNotification({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: record.id,
|
||||
title: "record reminder",
|
||||
message: "record message",
|
||||
});
|
||||
await egressStarted;
|
||||
await expect(
|
||||
getPrismaClient().notificationEgressAuthorization.count({
|
||||
where: {
|
||||
recordUserId: record.id,
|
||||
recipientUserId: recipient.id,
|
||||
channel: "NTFY",
|
||||
},
|
||||
}),
|
||||
).resolves.toBe(1);
|
||||
|
||||
const expiry = getPrismaClient().$transaction((tx) =>
|
||||
withManagedProfileLock(tx, record.id, () =>
|
||||
tx.accountGrant.update({
|
||||
where: { id: recipientGrant.id },
|
||||
data: { expiresAt: new Date() },
|
||||
}),
|
||||
),
|
||||
);
|
||||
await expectPromiseToSettle(expiry);
|
||||
|
||||
releaseEgress();
|
||||
await delivery;
|
||||
});
|
||||
|
||||
it("records a timed-out provider after the authorization claim has committed", async () => {
|
||||
const [record, recipient, reserveGuardian] = await Promise.all([
|
||||
createUser({ label: "record", locale: "en", managed: true }),
|
||||
createUser({ label: "recipient", locale: "en" }),
|
||||
createUser({ label: "reserve-guardian", locale: "de" }),
|
||||
]);
|
||||
await Promise.all([
|
||||
addGuardian(record.id, recipient.id),
|
||||
addGuardian(record.id, reserveGuardian.id),
|
||||
addChannel(recipient.id, "NTFY"),
|
||||
]);
|
||||
ntfyEgress.error = new Error("provider timeout");
|
||||
|
||||
await expect(
|
||||
dispatchNotification({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: record.id,
|
||||
title: "record reminder",
|
||||
message: "record message",
|
||||
}),
|
||||
).resolves.toEqual({
|
||||
dispatched: false,
|
||||
channelsAttempted: 1,
|
||||
channelsSucceeded: 0,
|
||||
});
|
||||
|
||||
await expect(
|
||||
getPrismaClient().notificationEgressAuthorization.findFirstOrThrow({
|
||||
where: {
|
||||
recordUserId: record.id,
|
||||
recipientUserId: recipient.id,
|
||||
channel: "NTFY",
|
||||
},
|
||||
select: { eventType: true, outcome: true, completedAt: true },
|
||||
}),
|
||||
).resolves.toEqual({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
outcome: "sender_threw",
|
||||
completedAt: expect.any(Date),
|
||||
});
|
||||
});
|
||||
|
||||
it("creates a new claim for each retry and stops after Guardian revocation", async () => {
|
||||
const [record, recipient, reserveGuardian] = await Promise.all([
|
||||
createUser({ label: "record", locale: "en", managed: true }),
|
||||
createUser({ label: "recipient", locale: "en" }),
|
||||
createUser({ label: "reserve-guardian", locale: "de" }),
|
||||
]);
|
||||
await Promise.all([
|
||||
addGuardian(record.id, recipient.id),
|
||||
addGuardian(record.id, reserveGuardian.id),
|
||||
addChannel(recipient.id, "NTFY"),
|
||||
]);
|
||||
const grant = await getPrismaClient().accountGrant.findFirstOrThrow({
|
||||
where: { grantorId: record.id, granteeId: recipient.id },
|
||||
select: { id: true },
|
||||
});
|
||||
const payload = {
|
||||
eventType: "MEDICATION_REMINDER" as const,
|
||||
userId: record.id,
|
||||
recordUserId: record.id,
|
||||
recipientUserId: recipient.id,
|
||||
title: "record reminder",
|
||||
message: "record message",
|
||||
};
|
||||
|
||||
ntfyEgress.error = new Error("provider timeout");
|
||||
await dispatchNotification(payload);
|
||||
await getPrismaClient().notificationChannel.update({
|
||||
where: { userId_type: { userId: recipient.id, type: "NTFY" } },
|
||||
data: { consecutiveFailures: 0, nextRetryAt: null },
|
||||
});
|
||||
ntfyEgress.error = null;
|
||||
ntfyEgress.outcome = { ok: false, reason: "provider_unavailable" };
|
||||
await dispatchNotification(payload);
|
||||
|
||||
const beforeRevocation =
|
||||
await getPrismaClient().notificationEgressAuthorization.findMany({
|
||||
where: {
|
||||
recordUserId: record.id,
|
||||
recipientUserId: recipient.id,
|
||||
channel: "NTFY",
|
||||
},
|
||||
select: { id: true, outcome: true },
|
||||
orderBy: { authorizedAt: "asc" },
|
||||
});
|
||||
expect(beforeRevocation).toEqual([
|
||||
{ id: expect.any(String), outcome: "sender_threw" },
|
||||
{ id: expect.any(String), outcome: "transient_failure" },
|
||||
]);
|
||||
expect(beforeRevocation[0]!.id).not.toBe(beforeRevocation[1]!.id);
|
||||
|
||||
await revokeGrant({ grantId: grant.id, grantorId: record.id });
|
||||
ntfyEgress.outcome = null;
|
||||
await expect(dispatchNotification(payload)).resolves.toEqual({
|
||||
dispatched: false,
|
||||
channelsAttempted: 0,
|
||||
channelsSucceeded: 0,
|
||||
});
|
||||
expect(sent.ntfy).toEqual([]);
|
||||
await expect(
|
||||
getPrismaClient().notificationEgressAuthorization.count({
|
||||
where: {
|
||||
recordUserId: record.id,
|
||||
recipientUserId: recipient.id,
|
||||
channel: "NTFY",
|
||||
},
|
||||
}),
|
||||
).resolves.toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
async function expectPromiseToSettle(promise: Promise<unknown>) {
|
||||
const settled = await Promise.race([
|
||||
promise.then(
|
||||
() => true,
|
||||
() => true,
|
||||
),
|
||||
new Promise<boolean>((resolve) => setTimeout(() => resolve(false), 250)),
|
||||
]);
|
||||
expect(settled).toBe(true);
|
||||
}
|
||||
@@ -1,10 +1,11 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
import { resolveNotificationDeliveryIdentity } from "@/lib/notifications/delivery-identity";
|
||||
import {
|
||||
claimNotificationEvent,
|
||||
} from "@/lib/notifications/reminder-dedup";
|
||||
claimManagedGuardianEgressAuthorization,
|
||||
resolveNotificationDeliveryIdentity,
|
||||
} from "@/lib/notifications/delivery-identity";
|
||||
import { claimNotificationEvent } from "@/lib/notifications/reminder-dedup";
|
||||
import { recordPushAttempt } from "@/lib/notifications/senders/push-attempt-record";
|
||||
|
||||
let sequence = 0;
|
||||
@@ -141,6 +142,132 @@ describe("notification attribution (real Postgres)", () => {
|
||||
).toBeNull();
|
||||
});
|
||||
|
||||
it("provides the created-at index used by global notification event retention", async () => {
|
||||
const indexes = await getPrismaClient().$queryRaw<
|
||||
Array<{ indexname: string }>
|
||||
>`
|
||||
SELECT indexname
|
||||
FROM pg_indexes
|
||||
WHERE schemaname = current_schema()
|
||||
AND tablename = 'notification_events'
|
||||
`;
|
||||
|
||||
expect(indexes.map(({ indexname }) => indexname)).toContain(
|
||||
"notification_events_created_at_idx",
|
||||
);
|
||||
});
|
||||
|
||||
it("refuses a revoked Guardian after identity preflight but before the final egress claim", async () => {
|
||||
const recordUser = await createUser("expiry-record", true);
|
||||
const recipientUser = await createUser("expiry-recipient");
|
||||
const grant = await createGuardianGrant({
|
||||
recordUserId: recordUser.id,
|
||||
recipientUserId: recipientUser.id,
|
||||
});
|
||||
const delivery = await resolveNotificationDeliveryIdentity({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: recordUser.id,
|
||||
recordUserId: recordUser.id,
|
||||
recipientUserId: recipientUser.id,
|
||||
title: "record reminder",
|
||||
message: "record message",
|
||||
});
|
||||
expect(delivery).toMatchObject({ managed: true });
|
||||
|
||||
await getPrismaClient().accountGrant.update({
|
||||
where: { id: grant.id },
|
||||
data: { revokedAt: new Date(), revokedBy: "GRANTOR" },
|
||||
});
|
||||
|
||||
await expect(
|
||||
claimManagedGuardianEgressAuthorization(
|
||||
delivery!,
|
||||
"NTFY",
|
||||
"MEDICATION_REMINDER",
|
||||
),
|
||||
).resolves.toBeNull();
|
||||
expect(
|
||||
await getPrismaClient().notificationEgressAuthorization.count(),
|
||||
).toBe(0);
|
||||
expect(await getPrismaClient().pushAttempt.count()).toBe(0);
|
||||
});
|
||||
|
||||
it("refuses an expired Guardian after identity preflight but before the final egress claim", async () => {
|
||||
const recordUser = await createUser("expiry-record", true);
|
||||
const recipientUser = await createUser("expiry-recipient");
|
||||
const grant = await createGuardianGrant({
|
||||
recordUserId: recordUser.id,
|
||||
recipientUserId: recipientUser.id,
|
||||
expiresAt: new Date(Date.now() + 60_000),
|
||||
});
|
||||
const delivery = await resolveNotificationDeliveryIdentity({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: recordUser.id,
|
||||
recordUserId: recordUser.id,
|
||||
recipientUserId: recipientUser.id,
|
||||
title: "record reminder",
|
||||
message: "record message",
|
||||
});
|
||||
expect(delivery).toMatchObject({ managed: true });
|
||||
|
||||
await getPrismaClient().accountGrant.update({
|
||||
where: { id: grant.id },
|
||||
data: { expiresAt: new Date(Date.now() - 1) },
|
||||
});
|
||||
|
||||
await expect(
|
||||
claimManagedGuardianEgressAuthorization(
|
||||
delivery!,
|
||||
"NTFY",
|
||||
"MEDICATION_REMINDER",
|
||||
),
|
||||
).resolves.toBeNull();
|
||||
expect(
|
||||
await getPrismaClient().notificationEgressAuthorization.count(),
|
||||
).toBe(0);
|
||||
expect(await getPrismaClient().pushAttempt.count()).toBe(0);
|
||||
});
|
||||
|
||||
it("uses database time for expiry at the final egress claim", async () => {
|
||||
const recordUser = await createUser("database-clock-record", true);
|
||||
const recipientUser = await createUser("database-clock-recipient");
|
||||
const actualNow = Date.now();
|
||||
await createGuardianGrant({
|
||||
recordUserId: recordUser.id,
|
||||
recipientUserId: recipientUser.id,
|
||||
expiresAt: new Date(actualNow - 1_000),
|
||||
});
|
||||
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(new Date(actualNow - 60_000));
|
||||
try {
|
||||
const delivery = await resolveNotificationDeliveryIdentity({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: recordUser.id,
|
||||
recordUserId: recordUser.id,
|
||||
recipientUserId: recipientUser.id,
|
||||
title: "record reminder",
|
||||
message: "record message",
|
||||
});
|
||||
expect(delivery).toMatchObject({ managed: true });
|
||||
|
||||
await expect(
|
||||
claimManagedGuardianEgressAuthorization(
|
||||
delivery!,
|
||||
"NTFY",
|
||||
"MEDICATION_REMINDER",
|
||||
),
|
||||
).resolves.toBeNull();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
|
||||
expect(
|
||||
await getPrismaClient().notificationEgressAuthorization.count(),
|
||||
).toBe(0);
|
||||
expect(await getPrismaClient().pushAttempt.count()).toBe(0);
|
||||
});
|
||||
|
||||
it("allows exactly one concurrent claim for a notification event", async () => {
|
||||
const recordUser = await createUser("claim-record", true);
|
||||
const client = getPrismaClient();
|
||||
|
||||
@@ -41,6 +41,10 @@ import {
|
||||
WIPE_OWNER_FIELDS,
|
||||
} from "@/lib/data-wipe/wipe-plan";
|
||||
import type { PrismaClient } from "@/generated/prisma/client";
|
||||
import {
|
||||
claimManagedGuardianEgressAuthorization,
|
||||
resolveNotificationDeliveryIdentity,
|
||||
} from "@/lib/notifications/delivery-identity";
|
||||
|
||||
vi.mock("next/headers", async () => {
|
||||
const { cookieJar, headerJar } = await import("./mock-next-headers");
|
||||
@@ -309,6 +313,18 @@ async function countRows(
|
||||
return Number(rows[0].n);
|
||||
}
|
||||
|
||||
async function managedProfileLockIsHeld(
|
||||
prisma: PrismaClient,
|
||||
profileId: string,
|
||||
): Promise<boolean> {
|
||||
const rows = await prisma.$queryRaw<Array<{ locked: boolean }>>`
|
||||
SELECT NOT pg_try_advisory_xact_lock(
|
||||
hashtextextended(${`managed-profile:${profileId}`}, 0)
|
||||
) AS locked
|
||||
`;
|
||||
return rows[0]?.locked === true;
|
||||
}
|
||||
|
||||
describe("DELETE /api/settings/data leaves nothing it promised to delete", () => {
|
||||
beforeEach(async () => {
|
||||
await truncateAllTables(getPrismaClient());
|
||||
@@ -467,6 +483,182 @@ describe("DELETE /api/settings/data leaves nothing it promised to delete", () =>
|
||||
).toEqual([]);
|
||||
});
|
||||
|
||||
it("refuses a wipe that would remove a managed profile's last Guardian", async () => {
|
||||
const prisma = getPrismaClient();
|
||||
const [guardian, profile] = await Promise.all([
|
||||
prisma.user.create({
|
||||
data: {
|
||||
username: "wipe-last-guardian",
|
||||
email: "wipe-last-guardian@example.test",
|
||||
},
|
||||
}),
|
||||
prisma.user.create({
|
||||
data: {
|
||||
username: "wipe-last-profile",
|
||||
email: "wipe-last-profile@example.test",
|
||||
managedProfileAt: new Date(),
|
||||
},
|
||||
}),
|
||||
]);
|
||||
await prisma.accountGrant.create({
|
||||
data: {
|
||||
grantorId: profile.id,
|
||||
granteeId: guardian.id,
|
||||
access: "MANAGE",
|
||||
acceptedAt: new Date(),
|
||||
},
|
||||
});
|
||||
await prisma.measurement.create({
|
||||
data: {
|
||||
userId: guardian.id,
|
||||
type: "WEIGHT",
|
||||
value: 80,
|
||||
unit: "kg",
|
||||
measuredAt: new Date(),
|
||||
},
|
||||
});
|
||||
const session = await prisma.session.create({
|
||||
data: { userId: guardian.id, expiresAt: new Date(Date.now() + 60_000) },
|
||||
});
|
||||
cookieJar.set("healthlog_session", session.id);
|
||||
|
||||
const { DELETE } = await import("@/app/api/settings/data/route");
|
||||
const response = await DELETE(
|
||||
new Request("http://localhost/api/settings/data", {
|
||||
method: "DELETE",
|
||||
body: JSON.stringify({ confirm: "DELETE" }),
|
||||
}) as never,
|
||||
);
|
||||
|
||||
expect(response.status).toBe(409);
|
||||
expect(
|
||||
await prisma.accountGrant.count({
|
||||
where: { grantorId: profile.id, granteeId: guardian.id },
|
||||
}),
|
||||
).toBe(1);
|
||||
expect(
|
||||
await prisma.measurement.count({ where: { userId: guardian.id } }),
|
||||
).toBe(1);
|
||||
});
|
||||
|
||||
it("makes a queued Guardian claim observe a completed data wipe", async () => {
|
||||
const prisma = getPrismaClient();
|
||||
const [guardian, reserveGuardian, profile] = await Promise.all([
|
||||
prisma.user.create({
|
||||
data: {
|
||||
username: "wipe-race-guardian",
|
||||
email: "wipe-race-guardian@example.test",
|
||||
},
|
||||
}),
|
||||
prisma.user.create({
|
||||
data: {
|
||||
username: "wipe-race-reserve",
|
||||
email: "wipe-race-reserve@example.test",
|
||||
},
|
||||
}),
|
||||
prisma.user.create({
|
||||
data: {
|
||||
username: "wipe-race-profile",
|
||||
email: "wipe-race-profile@example.test",
|
||||
managedProfileAt: new Date(),
|
||||
},
|
||||
}),
|
||||
]);
|
||||
await Promise.all([
|
||||
prisma.accountGrant.create({
|
||||
data: {
|
||||
grantorId: profile.id,
|
||||
granteeId: guardian.id,
|
||||
access: "MANAGE",
|
||||
acceptedAt: new Date(),
|
||||
},
|
||||
}),
|
||||
prisma.accountGrant.create({
|
||||
data: {
|
||||
grantorId: profile.id,
|
||||
granteeId: reserveGuardian.id,
|
||||
access: "MANAGE",
|
||||
acceptedAt: new Date(),
|
||||
},
|
||||
}),
|
||||
]);
|
||||
expect(
|
||||
await prisma.accountGrant.count({
|
||||
where: {
|
||||
grantorId: profile.id,
|
||||
access: "MANAGE",
|
||||
acceptedAt: { not: null },
|
||||
revokedAt: null,
|
||||
},
|
||||
}),
|
||||
).toBe(2);
|
||||
const measurement = await prisma.measurement.create({
|
||||
data: {
|
||||
userId: guardian.id,
|
||||
type: "WEIGHT",
|
||||
value: 80,
|
||||
unit: "kg",
|
||||
measuredAt: new Date(),
|
||||
},
|
||||
});
|
||||
const session = await prisma.session.create({
|
||||
data: { userId: guardian.id, expiresAt: new Date(Date.now() + 60_000) },
|
||||
});
|
||||
cookieJar.set("healthlog_session", session.id);
|
||||
const delivery = await resolveNotificationDeliveryIdentity({
|
||||
eventType: "MEDICATION_REMINDER",
|
||||
userId: profile.id,
|
||||
recordUserId: profile.id,
|
||||
recipientUserId: guardian.id,
|
||||
title: "record reminder",
|
||||
message: "record message",
|
||||
});
|
||||
expect(delivery).toMatchObject({ managed: true });
|
||||
|
||||
let releaseMeasurement!: () => void;
|
||||
let markMeasurementLocked!: () => void;
|
||||
const measurementLocked = new Promise<void>((resolve) => {
|
||||
markMeasurementLocked = resolve;
|
||||
});
|
||||
const holdMeasurement = new Promise<void>((resolve) => {
|
||||
releaseMeasurement = resolve;
|
||||
});
|
||||
const blocker = prisma.$transaction(async (tx) => {
|
||||
await tx.measurement.update({
|
||||
where: { id: measurement.id },
|
||||
data: { value: 81 },
|
||||
});
|
||||
markMeasurementLocked();
|
||||
await holdMeasurement;
|
||||
});
|
||||
await measurementLocked;
|
||||
|
||||
const { DELETE } = await import("@/app/api/settings/data/route");
|
||||
const wipe = DELETE(
|
||||
new Request("http://localhost/api/settings/data", {
|
||||
method: "DELETE",
|
||||
body: JSON.stringify({ confirm: "DELETE" }),
|
||||
}) as never,
|
||||
);
|
||||
await vi.waitFor(async () => {
|
||||
expect(await managedProfileLockIsHeld(prisma, profile.id)).toBe(true);
|
||||
});
|
||||
|
||||
const claim = claimManagedGuardianEgressAuthorization(
|
||||
delivery!,
|
||||
"NTFY",
|
||||
"MEDICATION_REMINDER",
|
||||
);
|
||||
await expectPromiseToRemainPending(claim);
|
||||
|
||||
releaseMeasurement();
|
||||
await blocker;
|
||||
expect((await wipe).status).toBe(200);
|
||||
await expect(claim).resolves.toBeNull();
|
||||
expect(await prisma.notificationEgressAuthorization.count()).toBe(0);
|
||||
expect(await prisma.pushAttempt.count()).toBe(0);
|
||||
});
|
||||
|
||||
it("refuses without the typed confirmation and leaves the record alone", async () => {
|
||||
const prisma = getPrismaClient();
|
||||
|
||||
@@ -542,3 +734,14 @@ describe("DELETE /api/settings/data leaves nothing it promised to delete", () =>
|
||||
expect(await resolveShareToken(rawToken)).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
async function expectPromiseToRemainPending(promise: Promise<unknown>) {
|
||||
const settled = await Promise.race([
|
||||
promise.then(
|
||||
() => true,
|
||||
() => true,
|
||||
),
|
||||
new Promise<boolean>((resolve) => setTimeout(() => resolve(false), 25)),
|
||||
]);
|
||||
expect(settled).toBe(false);
|
||||
}
|
||||
Reference in new issue
Block a user