Skip to content

fix(listener): repair unresolved-merge corruption blocking build/tests #757

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

Open
ibrahimMaryam-tech wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(listener): repair unresolved-merge corruption blocking build/tests #757

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
7 changes: 3 additions & 4 deletions listener/src/api/events-server.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -268,8 +268,7 @@ describe('POST /api/webhooks', () => {

expect(status).toBe(401);
expect((body as any).code).toBe('AUTH_INVALID_SIGNATURE');
expect((body as any).success).toBe(true);
expect((body as any).data.status).toBe('accepted');
expect((body as any).success).toBe(false);
});

it('rejects a timestamp-bound signature when the timestamp header is removed (anti-replay)', async () => {
Expand Down Expand Up @@ -412,13 +411,13 @@ describe('POST /api/webhooks', () => {
const payload = JSON.stringify({ event: 'test' });

server = await startServer({ ...BASE_OPTIONS, webhookSecrets: secrets });
await makePostRequest(server, '/api/webhooks', payload, {
const { body } = await makePostRequest(server, '/api/webhooks', payload, {
'X-Webhook-Key-Id': 'key-1',
});

expect(logger.warn).toHaveBeenCalled();
expect((body as any).success).toBe(false);
expect((body as any).error.message).toBe('Unknown key-id');
expect((body as any).error.message).toBe('Missing signature header');
});

it('returns 404 for POST to other paths', async () => {
Expand Down
139 changes: 31 additions & 108 deletions listener/src/api/events-server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,20 +12,11 @@ import { handleTemplateRoutes } from './template-routes';
import { sendOk, sendErr, sendJson, ErrorCode } from '../utils/response';
import { handleApiError, ApiError } from './error-handler';
import { applyRequestContext } from '../utils/request-id';
import { TemplateService } from '../services/template-service';
import { handleTemplateRoutes } from './template-routes';
import { NotificationHistoryService } from '../services/notification-history';
import { SearchSuggestionService } from '../services/search-suggestion';
import { NotificationSearchService } from '../services/notification-search-service';
import {
verifySignature,
extractSignature,
extractKeyId,
getSecretForKey,
collectRawBody,
extractTimestamp,
verifyWebhookRequest,
} from '../services/webhook-verifier';
import { collectRawBody, verifyWebhookRequest } from '../services/webhook-verifier';
import { IdempotencyKeyService, IdempotencyKeyReuseError } from '../services/idempotency-key-service';
import { WebhookSecret, RateLimitConfig, ContractConfig } from '../types';
import { RateLimiter } from './rate-limiter';
import { getDatabase } from '../database/database';
Expand All @@ -47,7 +38,7 @@ import {
serializeAuditRecord,
serializeTemplate,
} from './template-api';
import { CreateNotificationTemplateInputOld } from '../types/notification-template';
import { CreateNotificationTemplateInput } from '../types/notification-template';
import { BatchValidationService } from '../services/batch-validation-service';
import { handleArchiveRequest } from './archive-api';
import { ArchiveStore } from '../services/archive-store';
Expand All @@ -68,6 +59,8 @@ export interface EventsServerOptions {
webhookSecrets?: WebhookSecret[];
apiKeys?: Array<{ key: string; name?: string }>;
notificationAPI?: NotificationAPI | null;
/** Idempotency-Key replay protection for mutating endpoints (webhooks, schedule). */
idempotencyService?: IdempotencyKeyService | null;
templateService?: NotificationTemplateService | null;
/** Scheduler-scoped template service, used to render templates for scheduled notifications. */
schedulerTemplateService?: TemplateService | null;
Expand Down Expand Up @@ -628,25 +621,16 @@ export function createEventsServer(options: EventsServerOptions): http.Server {
// POST /api/webhooks
if (req.method === 'POST' && url.pathname === '/api/webhooks') {
const idempotencyKey = IdempotencyKeyService.extractKey(req.headers) ?? undefined;

const writeAuthFailure = (statusCode: number, message: string, code: string): void => {
res.writeHead(statusCode, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ success: false, error: { code, message }, code }));
};

collectRawBody(req).then(async (rawBody) => {
const sourceIp =
(req.headers['x-forwarded-for'] as string)?.split(',')[0]?.trim() ||
(req.socket?.remoteAddress as string | undefined);
collectRawBody(req).then((rawBody) => {
const signatureHeader = extractSignature(req.headers);
const keyId = extractKeyId(req.headers);

if (!signatureHeader) {
logger.warn('Webhook missing signature header', { requestId, correlationId });
sendErr(res, 401, 'Missing signature header', ErrorCode.UNAUTHORIZED);
return;
}

if (!keyId) {
logger.warn('Webhook missing key-id header', { requestId, correlationId });
sendErr(res, 401, 'Missing key-id header', ErrorCode.UNAUTHORIZED);
return;
}

const secrets = options.webhookSecrets ?? [];
const maxAgeSeconds = options.signatureExpirationSeconds ?? 300;
Expand All @@ -662,11 +646,12 @@ export function createEventsServer(options: EventsServerOptions): http.Server {
});

if (!auth.authenticated) {
res.writeHead(auth.statusCode, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: auth.message, code: auth.errorCode }));
if (!secret) {
logger.warn('Webhook unknown key-id', { requestId, correlationId, keyId });
sendErr(res, 401, 'Unknown key-id', ErrorCode.UNAUTHORIZED);
logger.warn('Webhook authentication failed', {
requestId,
correlationId,
code: auth.errorCode,
});
writeAuthFailure(auth.statusCode, auth.message, auth.errorCode);
return;
}

Expand Down Expand Up @@ -710,38 +695,16 @@ export function createEventsServer(options: EventsServerOptions): http.Server {
logger.warn('Webhook rejected: idempotency key reused with different body', {
requestId, correlationId, idempotencyKey,
});
res.writeHead(err.statusCode, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: err.message, code: err.code }));
if (timestampHeader) {
const timestamp = Array.isArray(timestampHeader) ? timestampHeader[0] : timestampHeader;
if (!isTimestampValid(timestamp, maxAgeSeconds)) {
logger.warn('Webhook request signature expired', { requestId, correlationId, keyId, timestamp });
sendErr(res, 401, 'Request signature expired', ErrorCode.UNAUTHORIZED);
writeAuthFailure(err.statusCode, err.message, err.code);
return;
}
throw err;
logger.error('Failed to process webhook', { requestId, correlationId, error: err });
sendErr(res, 500, 'Internal server error', ErrorCode.INTERNAL_ERROR);
}
}).catch((err) => {
if (err instanceof IdempotencyKeyReuseError) {
res.writeHead(err.statusCode, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: err.message, code: err.code }));
return;
}
logger.error('Failed to read webhook body', { requestId, correlationId, error: err instanceof Error ? err.message : String(err) });
res.writeHead(400, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: 'Failed to read request body', code: 'BODY_READ_FAILED' }));

if (!verifySignature(rawBody, signatureHeader, secret)) {
logger.warn('Webhook invalid signature', { requestId, correlationId, keyId });
sendErr(res, 401, 'Invalid signature', ErrorCode.UNAUTHORIZED);
return;
}

logger.info('Webhook received and verified', { requestId, correlationId, keyId });
sendOk(res, 202, { status: 'accepted' });
}).catch((err) => {
logger.error('Failed to read webhook body', { requestId, correlationId, error: err });
sendErr(res, 400, 'Failed to read request body', ErrorCode.BAD_REQUEST);
});
return;
}
Expand Down Expand Up @@ -833,15 +796,13 @@ export function createEventsServer(options: EventsServerOptions): http.Server {
if (!data.executeAt || !data.payload || !data.targetRecipient) {
res.writeHead(400, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: 'Missing required fields: executeAt, payload, targetRecipient', code: 'MISSING_FIELDS' }));
sendErr(res, 400, 'Missing required fields: executeAt, payload, targetRecipient', ErrorCode.BAD_REQUEST);
return;
}

const executeAt = new Date(data.executeAt);
if (isNaN(executeAt.getTime())) {
res.writeHead(400, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: 'executeAt is not a valid date', code: 'INVALID_DATE' }));
sendErr(res, 400, 'executeAt is not a valid date', ErrorCode.BAD_REQUEST);
return;
}

Expand Down Expand Up @@ -894,15 +855,7 @@ export function createEventsServer(options: EventsServerOptions): http.Server {
res.end(JSON.stringify({ error: error.message, code: error.code }));
return;
}
logger.error('Failed to schedule notification', {
error: error instanceof Error ? error.message : String(error),
requestId, correlationId,
});
res.writeHead(500, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: (error as Error).message, code: 'SCHEDULE_FAILED' }));
sendOk(res, 201, { id: notificationId });
logger.info('Notification scheduled via API', { requestId, correlationId, notificationId, executeAt: data.executeAt });
} catch (error) {

const anyError = error as any;
if (anyError?.name === 'PayloadTooLargeError') {
logger.warn('Payload too large', {
Expand All @@ -912,11 +865,17 @@ export function createEventsServer(options: EventsServerOptions): http.Server {
payloadSizeBytes: anyError.payloadSizeBytes,
maxSizeBytes: anyError.maxSizeBytes,
});
sendErr(res, 413, anyError.message, ErrorCode.PAYLOAD_TOO_LARGE);
res.writeHead(413, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: anyError.message, code: 'PAYLOAD_TOO_LARGE' }));
return;
}
logger.error('Failed to schedule notification', { error, requestId, correlationId });
sendErr(res, 500, (error as Error).message, ErrorCode.INTERNAL_ERROR);

logger.error('Failed to schedule notification', {
error: error instanceof Error ? error.message : String(error),
requestId, correlationId,
});
res.writeHead(500, { 'Content-Type': 'application/json' });
res.end(JSON.stringify({ error: (error as Error).message, code: 'SCHEDULE_FAILED' }));
}
});
return;
Expand Down Expand Up @@ -989,40 +948,6 @@ export function createEventsServer(options: EventsServerOptions): http.Server {
sendOk(res, 200, { failures: monitor.listFailures(limit), count: monitor.listFailures(limit).length });
return;
}
if (req.method === 'GET' && url.pathname === '/api/schedule/execution-metrics') {
if (!options.notificationAPI) {
sendErr(res, 503, 'Scheduler not enabled', ErrorCode.SERVICE_UNAVAILABLE);
return;
}

options.notificationAPI.getExecutionMetrics()
.then((metrics) => {
sendOk(res, 200, metrics);
})
.catch((error) => {
logger.error('Failed to get execution metrics', { error, requestId, correlationId });
handleApiError(res, error, requestId, correlationId);
});
return;
}

// GET /api/schedule/retry-distribution
if (req.method === 'GET' && url.pathname === '/api/schedule/retry-distribution') {
if (!options.notificationAPI) {
sendErr(res, 503, 'Scheduler not enabled', ErrorCode.SERVICE_UNAVAILABLE);
return;
}

options.notificationAPI.getRetryDistribution()
.then((distribution) => {
sendOk(res, 200, distribution);
})
.catch((error) => {
logger.error('Failed to get retry distribution', { error, requestId, correlationId });
handleApiError(res, error, requestId, correlationId);
});
return;
}

// GET /api/schedule/retry-statistics
if (req.method === 'GET' && url.pathname === '/api/schedule/retry-statistics') {
Expand Down Expand Up @@ -1140,7 +1065,6 @@ export function createEventsServer(options: EventsServerOptions): http.Server {
.then((result) => {
sendJson(res, 200, result);
logger.info('GET /api/notifications/history complete', {
requestId, total: result.total, durationMs: Date.now() - startTime,
requestId,
correlationId,
total: result.total,
Expand Down Expand Up @@ -1185,7 +1109,6 @@ export function createEventsServer(options: EventsServerOptions): http.Server {
sortBy,
});

notificationSearchService.search({ q, sender, txHash, eventId, status, type, startDate, endDate, limit, offset })
notificationSearchService.search({
q,
sender,
Expand Down Expand Up @@ -1384,7 +1307,7 @@ export function createEventsServer(options: EventsServerOptions): http.Server {
req.on('end', () => {
void (async () => {
try {
const parsed = JSON.parse(body) as CreateNotificationTemplateInputOld;
const parsed = JSON.parse(body) as CreateNotificationTemplateInput;
if (!parsed?.id || !parsed?.name || !parsed?.type || !parsed?.body) {
sendErr(res, 400, 'Invalid body: id, name, type, and body are required', ErrorCode.BAD_REQUEST);
return;
Expand Down
Loading