Skip to content

fix(api): gate /alerts/ingest on alerts:ingest API key scope #4208

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dipexplorer merged 1 commit into from
Aug 8, 2026
+202 −151
Merged

fix(api): gate /alerts/ingest on alerts:ingest API key scope #4208

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions apps/api/src/middleware/apiKeyAuth.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,26 @@ export interface ApiKeyRequest extends Request {
apiKey?: ApiKeyInfo;
}

/**
* Route middleware that requires the authenticated API key to carry a specific
* scope. Must run after `requireApiKey` so `req.apiKey` is populated. Returns
* 403 when the key is valid but lacks the required scope, so a narrowly-scoped
* key can never act with elevated privileges.
*/
export const requireApiKeyScope = (scope: string) => {
return (req: ApiKeyRequest, res: Response, next: NextFunction) => {
if (!req.apiKey) {
res.status(401).json({ error: "Unauthorized" });
return;
}
if (!req.apiKey.scopes?.includes(scope)) {
res.status(403).json({ error: "Forbidden: API key lacks required scope" });
return;
}
next();
};
};

export const requireApiKey = async (req: ApiKeyRequest, res: Response, next: NextFunction) => {
const apiKey = req.headers["x-api-secret"] as string | undefined;

Expand Down
308 changes: 157 additions & 151 deletions apps/api/src/routes/alerts.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import { triggerRecallAlert } from "../services/notifications";
import { validateMedicineStatus, getValidStatusList } from "../validators/medicine.validator";
import { escapeIlike } from "../utils/db";
import { requireApiKey, ApiKeyRequest } from "../middleware/apiKeyAuth";
import { requireApiKey, requireApiKeyScope, ApiKeyRequest } from "../middleware/apiKeyAuth";
import logger from "../utils/logger";
import { redisClient } from "../utils/redis";
import { KEY_PREFIXES, invalidateCacheByPattern } from "../services/cache.service";
Expand Down Expand Up @@ -191,183 +191,189 @@
* POST /api/v1/alerts/ingest
* Protected endpoint to ingest parsed CDSCO alerts from the ML agent.
*/
alertsRouter.post("/ingest", requireApiKey, limiter, async (req: ApiKeyRequest, res: Response) => {
const ingestSchema = z
.object({
alerts: AlertsArraySchema,
})
.strict();

const parseResult = ingestSchema.safeParse(req.body);
if (!parseResult.success) {
res.status(400).json({
error: "Invalid payload schema or unknown fields",
details: parseResult.error,
});
return;
}

const validatedAlerts = parseResult.data.alerts;

try {
// 2. Upsert alerts — ON CONFLICT DO NOTHING prevents duplicate rows
// when concurrent scraper instances race past the pre-check in deduplicate_alerts().
const { data: insertedAlerts, error: insertError } = await supabase
.from("drug_alerts")
.upsert(validatedAlerts, {
onConflict: "batch_number,source_url",
ignoreDuplicates: true,
alertsRouter.post(
"/ingest",
requireApiKey,
requireApiKeyScope("alerts:ingest"),
limiter,
async (req: ApiKeyRequest, res: Response) => {

Check failure on line 199 in apps/api/src/routes/alerts.ts

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this function to reduce its Cognitive Complexity from 37 to the 15 allowed. 


See more on https://sonarcloud.io/project/issues?id=RatLoopz_sahidawa-india&issues=AZ_cVWkdHpVEGdmuKVYx&open=AZ_cVWkdHpVEGdmuKVYx&pullRequest=4208
const ingestSchema = z
.object({
alerts: AlertsArraySchema,
})
.select();

if (insertError) {
logger.error("Error inserting alerts", { error: insertError });
res.status(500).json({ error: "Database error inserting alerts" });
return;
}
.strict();

// 3. Update medicines table based on matched batches
const medicineStatus = "recalled";
if (!validateMedicineStatus(medicineStatus)) {
const parseResult = ingestSchema.safeParse(req.body);
if (!parseResult.success) {
res.status(400).json({
error: `Invalid medicine status. Valid values are: ${getValidStatusList()}`,
error: "Invalid payload schema or unknown fields",
details: parseResult.error,
});
return;
}

// Batch the medicine status updates to avoid O(N) individual UPDATE queries.
// Alerts are grouped into two buckets by their secondary discriminator:
// - byManufacturer: alerts that have a manufacturer field
// - byBrandName: alerts that have only a brand name (no manufacturer)
// Each bucket is resolved in a single UPDATE ... WHERE batch_number IN (...)
// query, capping the total number of DB round-trips at 2 regardless of N.
const byManufacturer = new Map<string, string[]>(); // manufacturer -> batch_numbers[]
const byBrandName = new Map<string, string[]>(); // brand_name -> batch_numbers[]
const noBatchAlerts: typeof validatedAlerts = [];

for (const alert of validatedAlerts) {
if (!alert.batch_number) {
noBatchAlerts.push(alert);
continue;
const validatedAlerts = parseResult.data.alerts;

try {
// 2. Upsert alerts — ON CONFLICT DO NOTHING prevents duplicate rows
// when concurrent scraper instances race past the pre-check in deduplicate_alerts().
const { data: insertedAlerts, error: insertError } = await supabase
.from("drug_alerts")
.upsert(validatedAlerts, {
onConflict: "batch_number,source_url",
ignoreDuplicates: true,
})
.select();

if (insertError) {
logger.error("Error inserting alerts", { error: insertError });
res.status(500).json({ error: "Database error inserting alerts" });
return;
}

// 3. Update medicines table based on matched batches
const medicineStatus = "recalled";
if (!validateMedicineStatus(medicineStatus)) {
res.status(400).json({
error: `Invalid medicine status. Valid values are: ${getValidStatusList()}`,
});
return;
}
if (alert.manufacturer) {
if (!byManufacturer.has(alert.manufacturer)) {
byManufacturer.set(alert.manufacturer, []);

// Batch the medicine status updates to avoid O(N) individual UPDATE queries.
// Alerts are grouped into two buckets by their secondary discriminator:
// - byManufacturer: alerts that have a manufacturer field
// - byBrandName: alerts that have only a brand name (no manufacturer)
// Each bucket is resolved in a single UPDATE ... WHERE batch_number IN (...)
// query, capping the total number of DB round-trips at 2 regardless of N.
const byManufacturer = new Map<string, string[]>(); // manufacturer -> batch_numbers[]
const byBrandName = new Map<string, string[]>(); // brand_name -> batch_numbers[]
const noBatchAlerts: typeof validatedAlerts = [];

for (const alert of validatedAlerts) {
if (!alert.batch_number) {
noBatchAlerts.push(alert);
continue;
}
byManufacturer.get(alert.manufacturer)!.push(alert.batch_number);
} else if (alert.reported_brand_name) {
if (!byBrandName.has(alert.reported_brand_name)) {
byBrandName.set(alert.reported_brand_name, []);
if (alert.manufacturer) {
if (!byManufacturer.has(alert.manufacturer)) {
byManufacturer.set(alert.manufacturer, []);
}
byManufacturer.get(alert.manufacturer)!.push(alert.batch_number);
} else if (alert.reported_brand_name) {
if (!byBrandName.has(alert.reported_brand_name)) {
byBrandName.set(alert.reported_brand_name, []);
}
byBrandName.get(alert.reported_brand_name)!.push(alert.batch_number);
}
byBrandName.get(alert.reported_brand_name)!.push(alert.batch_number);
}
}

const batchUpdatePromises: Promise<unknown>[] = [];

for (const [manufacturer, batchNumbers] of byManufacturer) {
batchUpdatePromises.push(
Promise.resolve(
supabase
.from("medicines")
.update({ status: medicineStatus, is_counterfeit_alert: true })
.in("batch_number", batchNumbers)
.eq("manufacturer", manufacturer)
)
);
}
const batchUpdatePromises: Promise<unknown>[] = [];

for (const [manufacturer, batchNumbers] of byManufacturer) {
batchUpdatePromises.push(
Promise.resolve(
supabase
.from("medicines")
.update({ status: medicineStatus, is_counterfeit_alert: true })
.in("batch_number", batchNumbers)
.eq("manufacturer", manufacturer)
)
);
}

for (const [brandName, batchNumbers] of byBrandName) {
batchUpdatePromises.push(
Promise.resolve(
supabase
.from("medicines")
.update({ status: medicineStatus, is_counterfeit_alert: true })
.in("batch_number", batchNumbers)
.eq("brand_name", brandName)
)
);
}
for (const [brandName, batchNumbers] of byBrandName) {
batchUpdatePromises.push(
Promise.resolve(
supabase
.from("medicines")
.update({ status: medicineStatus, is_counterfeit_alert: true })
.in("batch_number", batchNumbers)
.eq("brand_name", brandName)
)
);
}

await Promise.all(batchUpdatePromises);
await Promise.all(batchUpdatePromises);

// 3.5 Invalidate the cache for the updated batch numbers
// Use pattern-based invalidation (drug:batch:<batch>*) to delete both batch-only
// and composite keys (drug:batch:<batch>|<barcode>|<brand>). This prevents stale
// counterfeit/recall data from being served from cache after an alert is ingested.
const batchNumbersToInvalidate = validatedAlerts
.map((alert) => alert.batch_number)
.filter(Boolean) as string[];
// 3.5 Invalidate the cache for the updated batch numbers
// Use pattern-based invalidation (drug:batch:<batch>*) to delete both batch-only
// and composite keys (drug:batch:<batch>|<barcode>|<brand>). This prevents stale
// counterfeit/recall data from being served from cache after an alert is ingested.
const batchNumbersToInvalidate = validatedAlerts
.map((alert) => alert.batch_number)
.filter(Boolean) as string[];

if (batchNumbersToInvalidate.length > 0 && redisClient.isOpen) {
try {
for (const batch of batchNumbersToInvalidate) {
await invalidateCacheByPattern(`${KEY_PREFIXES.DRUG_CACHE}${batch}*`);
if (batchNumbersToInvalidate.length > 0 && redisClient.isOpen) {
try {
for (const batch of batchNumbersToInvalidate) {
await invalidateCacheByPattern(`${KEY_PREFIXES.DRUG_CACHE}${batch}*`);
}
} catch (err) {
logger.error({
message: "Failed to invalidate cache for alert batches",
error: err,
});
}
} catch (err) {
logger.error({
message: "Failed to invalidate cache for alert batches",
error: err,
});
}
}

// NEW: Invalidate all paginated list caches since new alerts change page 1 results
if (redisClient.isOpen) {
try {
const listKeys: string[] = [];
for await (const key of redisClient.scanIterator({
MATCH: "alerts:list:*",
})) {
if (Array.isArray(key)) {
listKeys.push(...(key as string[]));
} else {
listKeys.push(key as unknown as string);
// NEW: Invalidate all paginated list caches since new alerts change page 1 results
if (redisClient.isOpen) {
try {
const listKeys: string[] = [];
for await (const key of redisClient.scanIterator({
MATCH: "alerts:list:*",
})) {
if (Array.isArray(key)) {
listKeys.push(...(key as string[]));
} else {
listKeys.push(key as unknown as string);
}
}
if (listKeys.length > 0) {
await redisClient.del(listKeys);
}
} catch (err) {
logger.error({
message: "Failed to invalidate alerts:list cache",
error: err,
});
}
if (listKeys.length > 0) {
await redisClient.del(listKeys);
}
} catch (err) {
logger.error({
message: "Failed to invalidate alerts:list cache",
error: err,
});
}
}

// 4. Dispatch Web Push Notifications using shared service
if (insertedAlerts && insertedAlerts.length > 0) {
const pushPromises = insertedAlerts.map((alert) => {
return triggerRecallAlert({
id: alert.id ? String(alert.id) : "unknown",
medicineName: alert.reported_brand_name || "Unknown Medicine",
batchNumber: alert.batch_number,
manufacturer: alert.manufacturer,
reason: `Alert of type ${alert.alert_type || "NSQ"} in ${alert.state || "Unknown region"}`,
severity: "high",
source: "CDSCO Live Feed",
recalledAt: alert.reported_at || new Date().toISOString(),
// 4. Dispatch Web Push Notifications using shared service
if (insertedAlerts && insertedAlerts.length > 0) {
const pushPromises = insertedAlerts.map((alert) => {
return triggerRecallAlert({
id: alert.id ? String(alert.id) : "unknown",
medicineName: alert.reported_brand_name || "Unknown Medicine",
batchNumber: alert.batch_number,
manufacturer: alert.manufacturer,
reason: `Alert of type ${alert.alert_type || "NSQ"} in ${alert.state || "Unknown region"}`,
severity: "high",
source: "CDSCO Live Feed",
recalledAt: alert.reported_at || new Date().toISOString(),
});
});
});
await Promise.all(pushPromises);
}
await Promise.all(pushPromises);
}

logger.info("Alerts ingested successfully", {
caller: req.apiKey?.userId,
count: insertedAlerts?.length,
});
logger.info("Alerts ingested successfully", {
caller: req.apiKey?.userId,
count: insertedAlerts?.length,
});

res.status(200).json({
success: true,
message: "Alerts ingested and notifications dispatched",
inserted: insertedAlerts?.length,
});
} catch (error) {
logger.error("Unexpected error in /ingest", { error, caller: req.apiKey?.userId });
res.status(500).json({ error: "Internal server error" });
res.status(200).json({
success: true,
message: "Alerts ingested and notifications dispatched",
inserted: insertedAlerts?.length,
});
} catch (error) {
logger.error("Unexpected error in /ingest", { error, caller: req.apiKey?.userId });
res.status(500).json({ error: "Internal server error" });
}
}
});
);

/**
* PATCH /api/v1/alerts/:id/snooze
Expand Down
Loading
Loading