Error: {error}
@@ -284,15 +304,33 @@ export function EventExplorerPage() {
)}
+ {refreshError && filteredEvents.length > 0 && (
+
+
+ Refresh Error: {refreshError} — existing events are still displayed.
+
+
+
+ )}
+
Showing {fromIndex.toLocaleString()}–{toIndex.toLocaleString()} of{' '}
{filteredEvents.length.toLocaleString()} events
- {isLoading &&
Loading events…
}
+ {(isLoading || isRefreshing) && (
+
+ {isRefreshing ? 'Refreshing events…' : 'Loading events…'}
+
+ )}
- {isLoading ? (
+ {isLoading && filteredEvents.length === 0 ? (
) : currentPageEvents.length > 0 ? (
= 0
+ ) {
+ latencySum += r.deliveryLatencyMs;
+ latencyCount++;
+ }
}
const total = success + failure + retry + skipped;
const terminal = success + failure;
const successRate = terminal > 0 ? success / terminal : 0;
const averageDurationMs = durationCount > 0 ? durationSum / durationCount : 0;
+ const averageDeliveryLatencyMs = latencyCount > 0 ? latencySum / latencyCount : 0;
return {
total,
@@ -279,12 +294,11 @@ export class NotificationAnalyticsAggregator {
skipped,
successRate,
averageDurationMs,
+ averageDeliveryLatencyMs,
};
}
- private computeByType(
- visible: AnalyticsDeliveryRecord[],
- ): AnalyticsByTypeSnapshot[] {
+ private computeByType(visible: AnalyticsDeliveryRecord[]): AnalyticsByTypeSnapshot[] {
const map = new Map();
for (const r of visible) {
const entry = map.get(r.notificationType) ?? { total: 0, success: 0, failure: 0 };
@@ -309,9 +323,7 @@ export class NotificationAnalyticsAggregator {
return result;
}
- private computeByContract(
- visible: AnalyticsDeliveryRecord[],
- ): AnalyticsByContractSnapshot[] {
+ private computeByContract(visible: AnalyticsDeliveryRecord[]): AnalyticsByContractSnapshot[] {
const map = new Map();
for (const r of visible) {
if (!r.contractAddress) continue;
@@ -345,8 +357,7 @@ export class NotificationAnalyticsAggregator {
now: number,
): AnalyticsBucketSnapshot[] {
const newestBucketStart = Math.floor(now / this.bucketSizeMs) * this.bucketSizeMs;
- const oldestBucketStart =
- newestBucketStart - (this.maxBuckets - 1) * this.bucketSizeMs;
+ const oldestBucketStart = newestBucketStart - (this.maxBuckets - 1) * this.bucketSizeMs;
const buckets: AnalyticsBucketSnapshot[] = [];
const indexByStart = new Map();
@@ -360,6 +371,7 @@ export class NotificationAnalyticsAggregator {
retry: 0,
skipped: 0,
averageDurationMs: 0,
+ averageDeliveryLatencyMs: 0,
};
indexByStart.set(t, buckets.length);
buckets.push(snapshot);
@@ -369,9 +381,12 @@ export class NotificationAnalyticsAggregator {
let durationCountBucket = 0;
let durationBucketIdx = -1;
+ let latencySumBucket = 0;
+ let latencyCountBucket = 0;
+ let latencyBucketIdx = -1;
+
for (const r of visible) {
- const bucketStart =
- Math.floor(r.timestamp / this.bucketSizeMs) * this.bucketSizeMs;
+ const bucketStart = Math.floor(r.timestamp / this.bucketSizeMs) * this.bucketSizeMs;
const idx = indexByStart.get(bucketStart);
if (idx === undefined) continue;
@@ -398,14 +413,30 @@ export class NotificationAnalyticsAggregator {
averageDurationMs: durationSumBucket / durationCountBucket,
};
}
+
+ if (
+ r.outcome === 'success' &&
+ r.deliveryLatencyMs !== undefined &&
+ r.deliveryLatencyMs >= 0
+ ) {
+ if (idx !== latencyBucketIdx) {
+ latencySumBucket = 0;
+ latencyCountBucket = 0;
+ latencyBucketIdx = idx;
+ }
+ latencySumBucket += r.deliveryLatencyMs;
+ latencyCountBucket++;
+ buckets[idx] = {
+ ...buckets[idx],
+ averageDeliveryLatencyMs: latencySumBucket / latencyCountBucket,
+ };
+ }
}
return buckets;
}
- private computeErrorBreakdown(
- visible: AnalyticsDeliveryRecord[],
- ): Record {
+ private computeErrorBreakdown(visible: AnalyticsDeliveryRecord[]): Record {
const counts = new Map();
for (const r of visible) {
if (r.outcome !== 'failure') continue;
diff --git a/listener/src/services/notification-scheduler.ts b/listener/src/services/notification-scheduler.ts
index 8d0cda9b..b0c7b882 100644
--- a/listener/src/services/notification-scheduler.ts
+++ b/listener/src/services/notification-scheduler.ts
@@ -10,6 +10,10 @@ import { getWorkerManager } from './worker-manager';
import { getJobMonitor } from './job-monitor';
import { ProviderRegistry, getProviderRegistry } from './provider-registry';
import { verifyPayloadIntegrity } from '../utils/payload-integrity';
+import {
+ getNotificationAnalyticsAggregator,
+ NotificationAnalyticsAggregator,
+} from './notification-analytics-aggregator';
/**
* Background scheduler that processes scheduled notifications
@@ -32,13 +36,14 @@ export class NotificationScheduler {
* When not supplied the module-level singleton is used.
*/
private providerRegistry: ProviderRegistry;
+ private analytics: NotificationAnalyticsAggregator;
constructor(
repository: ScheduledNotificationRepository,
config: SchedulerConfig,
discordService?: DiscordNotificationService | null,
batchValidator?: BatchValidationService,
- providerRegistry?: ProviderRegistry
+ providerRegistry?: ProviderRegistry,
) {
this.repository = repository;
this.config = { retryDelayMs: 5_000, ...config };
@@ -46,6 +51,7 @@ export class NotificationScheduler {
this.processorId = config.processorId || uuidv4();
this.batchValidator = batchValidator ?? new BatchValidationService();
this.providerRegistry = providerRegistry ?? getProviderRegistry();
+ this.analytics = getNotificationAnalyticsAggregator();
}
/**
@@ -126,7 +132,7 @@ export class NotificationScheduler {
this.processorId,
this.config.lockTimeoutMs,
this.config.batchSize,
- requestId
+ requestId,
);
if (notifications.length === 0) {
@@ -140,7 +146,7 @@ export class NotificationScheduler {
}
const batchRejection = this.batchValidator.rejectIfInvalid(
- this.toValidationBatch(notifications)
+ this.toValidationBatch(notifications),
);
if (batchRejection) {
@@ -153,9 +159,11 @@ export class NotificationScheduler {
for (const notification of notifications) {
await this.repository.markAsFailedOrRetry(
notification.id!,
- new Error(`Batch validation failed: ${batchRejection.errors.map((e) => e.message).join('; ')}`),
+ new Error(
+ `Batch validation failed: ${batchRejection.errors.map((e) => e.message).join('; ')}`,
+ ),
notification.retryCount,
- notification.maxRetries
+ notification.maxRetries,
);
}
return;
@@ -180,7 +188,7 @@ export class NotificationScheduler {
notification.id!,
new Error('Scheduler shutting down'),
notification.retryCount,
- notification.maxRetries
+ notification.maxRetries,
);
}
return;
@@ -197,7 +205,7 @@ export class NotificationScheduler {
notification.id!,
new Error('Scheduler shutting down'),
notification.retryCount,
- notification.maxRetries
+ notification.maxRetries,
);
continue;
}
@@ -237,7 +245,7 @@ export class NotificationScheduler {
private async processNotification(
notification: ScheduledNotification,
requestId: string,
- jobId?: string
+ jobId?: string,
): Promise {
const startTime = Date.now();
const executionAttempt = notification.retryCount + 1;
@@ -268,7 +276,7 @@ export class NotificationScheduler {
notification.id!,
new Error('Not yet due for execution'),
notification.retryCount,
- notification.maxRetries
+ notification.maxRetries,
);
if (jobId) {
jobMonitor.failJob(jobId, 'Not yet due for execution', {
@@ -296,7 +304,9 @@ export class NotificationScheduler {
requestId,
id: notification.id,
});
- } else if (!verifyPayloadIntegrity(notification.payload, notification.payloadHash, secret)) {
+ } else if (
+ !verifyPayloadIntegrity(notification.payload, notification.payloadHash, secret)
+ ) {
logger.error('Payload integrity verification failed — rejecting notification', {
requestId,
id: notification.id,
@@ -306,7 +316,7 @@ export class NotificationScheduler {
notification.id!,
new Error('Payload integrity check failed: hash mismatch'),
notification.maxRetries, // exhaust retries — don't retry a tampered payload
- notification.maxRetries
+ notification.maxRetries,
);
if (jobId) {
jobMonitor.failJob(jobId, 'Payload integrity check failed: hash mismatch', {
@@ -339,11 +349,25 @@ export class NotificationScheduler {
});
}
+ const deliveryLatencyMs = notification.createdAt
+ ? Date.now() - new Date(notification.createdAt).getTime()
+ : undefined;
+
+ this.analytics.record({
+ notificationType: notification.notificationType,
+ contractAddress: notification.contractAddress ?? undefined,
+ outcome: 'success',
+ durationMs,
+ deliveryLatencyMs,
+ timestamp: Date.now(),
+ });
+
logger.info('Notification delivered successfully', {
requestId,
id: notification.id,
type: notification.notificationType,
durationMs,
+ deliveryLatencyMs,
});
} else {
throw new Error('Notification delivery returned false');
@@ -365,6 +389,15 @@ export class NotificationScheduler {
});
}
+ this.analytics.record({
+ notificationType: notification.notificationType,
+ contractAddress: notification.contractAddress ?? undefined,
+ outcome: 'failure',
+ durationMs,
+ errorReason: (error as Error).message,
+ timestamp: Date.now(),
+ });
+
const willRetry = notification.retryCount + 1 < notification.maxRetries;
const nextRetryAt = willRetry
? new Date(Date.now() + (this.config.retryDelayMs ?? 5_000))
@@ -375,7 +408,7 @@ export class NotificationScheduler {
error as Error,
notification.retryCount,
notification.maxRetries,
- nextRetryAt
+ nextRetryAt,
);
await this.repository.logExecution({
@@ -403,7 +436,7 @@ export class NotificationScheduler {
*/
private async executeNotification(
notification: ScheduledNotification,
- requestId: string
+ requestId: string,
): Promise {
const payload = JSON.parse(notification.payload);
const type = notification.notificationType;
@@ -442,33 +475,33 @@ export class NotificationScheduler {
case 'discord':
if (!this.discordService) {
throw new Error(
- 'Discord service not configured and no Discord provider registered in the registry'
+ 'Discord service not configured and no Discord provider registered in the registry',
);
}
return await this.discordService.sendEventNotification(
payload.event,
payload.contractConfig,
- `scheduler-${notification.id}-${requestId}`
+ `scheduler-${notification.id}-${requestId}`,
);
case 'webhook':
throw new Error(
- 'Webhook delivery not yet implemented. Register a WebhookNotificationProvider in the ProviderRegistry.'
+ 'Webhook delivery not yet implemented. Register a WebhookNotificationProvider in the ProviderRegistry.',
);
case 'email':
throw new Error(
- 'Email delivery not yet implemented. Register an email NotificationProvider in the ProviderRegistry.'
+ 'Email delivery not yet implemented. Register an email NotificationProvider in the ProviderRegistry.',
);
case 'sms':
throw new Error(
- 'SMS delivery not yet implemented. Register an SMS NotificationProvider in the ProviderRegistry.'
+ 'SMS delivery not yet implemented. Register an SMS NotificationProvider in the ProviderRegistry.',
);
default:
throw new Error(
- `Unsupported notification type: "${type}". Register a provider for this type in the ProviderRegistry.`
+ `Unsupported notification type: "${type}". Register a provider for this type in the ProviderRegistry.`,
);
}
}
diff --git a/listener/src/test-utils/event-fixtures.test.ts b/listener/src/test-utils/event-fixtures.test.ts
new file mode 100644
index 00000000..701835dc
--- /dev/null
+++ b/listener/src/test-utils/event-fixtures.test.ts
@@ -0,0 +1,38 @@
+import { EventFixtures } from './event-fixtures';
+import { xdr } from '@stellar/stellar-sdk';
+
+describe('EventFixtures', () => {
+ it('generates a valid event', () => {
+ const event = EventFixtures.valid();
+ expect(event.id).toBe('evt-valid-1');
+ expect(event.type).toBe('contract');
+ expect(event.topic).toBeDefined();
+ expect(event.value).toBeDefined();
+ });
+
+ it('generates duplicate events', () => {
+ const events = EventFixtures.duplicate();
+ expect(events.length).toBe(2);
+ expect(events[0].id).toBe(events[1].id);
+ expect(events[0].txHash).toBe(events[1].txHash);
+ });
+
+ it('generates missing fields event', () => {
+ const event = EventFixtures.missingFields();
+ expect(event.type).toBe('contract');
+ expect(event.id).toBeUndefined();
+ });
+
+ it('generates unsupported version event', () => {
+ const event = EventFixtures.unsupportedVersion();
+ const val = event.value.map();
+ expect(val).toBeDefined();
+ const versionEntry = val?.find((entry) => entry.key().sym().toString() === 'version');
+ expect(versionEntry?.val().u32()).toBe(999);
+ });
+
+ it('generates malformed payload event', () => {
+ const event = EventFixtures.malformedPayload();
+ expect(event.ledger).toBe(-1);
+ });
+});
diff --git a/listener/src/test-utils/event-fixtures.ts b/listener/src/test-utils/event-fixtures.ts
new file mode 100644
index 00000000..f59e8614
--- /dev/null
+++ b/listener/src/test-utils/event-fixtures.ts
@@ -0,0 +1,54 @@
+import * as StellarSDK from '@stellar/stellar-sdk';
+import { xdr } from '@stellar/stellar-sdk';
+
+const defaultContract = 'CBIELTK6YBZJU5UP2WWQEUCYKLPU6AUNZ2BQ4WWFEIE3USCIU6KPNBAM';
+
+export const EventFixtures = {
+ valid: (
+ overrides: Partial = {},
+ ): StellarSDK.rpc.Api.EventResponse =>
+ ({
+ id: 'evt-valid-1',
+ type: 'contract',
+ ledger: 1000,
+ ledgerClosedAt: '2026-06-22T00:00:00Z',
+ transactionIndex: 0,
+ operationIndex: 0,
+ inSuccessfulContractCall: true,
+ txHash: 'tx-valid-abc',
+ topic: [xdr.scvSymbol('test_event')],
+ value: xdr.scvString('valid payload'),
+ contractId: { contractId: () => defaultContract } as any, // mock for contractId if needed
+ ...overrides,
+ }) as StellarSDK.rpc.Api.EventResponse,
+
+ duplicate: (): StellarSDK.rpc.Api.EventResponse[] => {
+ const base = EventFixtures.valid({ id: 'evt-dup-1', txHash: 'tx-dup-1' });
+ return [base, { ...base }];
+ },
+
+ missingFields: (): Partial => ({
+ // Missing id, topic, value, etc.
+ type: 'contract',
+ ledger: 1001,
+ inSuccessfulContractCall: true,
+ }),
+
+ unsupportedVersion: (): StellarSDK.rpc.Api.EventResponse =>
+ EventFixtures.valid({
+ id: 'evt-unsupported-1',
+ value: xdr.scvMap([
+ new xdr.ScMapEntry({
+ key: xdr.scvSymbol('version'),
+ val: xdr.scvU32(999), // Unsupported version
+ }),
+ ]),
+ }),
+
+ malformedPayload: (): StellarSDK.rpc.Api.EventResponse =>
+ EventFixtures.valid({
+ id: 'evt-malformed-1',
+ // validateEventPayload explicitly rejects ledger < 0 as malformed
+ ledger: -1,
+ }),
+};