Skip to content

Commit 31d7e45

Browse files
coodoscoodos
committed
feat(evault-core): replace webhook fanout with AaaS ingest
evault-core no longer queries the registry and fans out webhooks to every platform. getActivePlatforms and deliverWebhooks are removed; a single notifyAwareness POST forwards each awareness packet to AWARENESS_SERVICE_URL/ ingest, authenticated by a shared secret. All five mutation call sites (create, update, bulk-create, binding document create/sign) are updated. The requesting platform is passed through to AaaS so it can skip delivering a packet back to its origin, preserving the ping-pong guard the old fanout had.
1 parent 6452511 commit 31d7e45

3 files changed

Lines changed: 96 additions & 140 deletions

File tree

‎infrastructure/evault-core/src/core/protocol/graphql-server.ts‎

Lines changed: 69 additions & 139 deletions
Original file line numberDiff line numberDiff line change
@@ -62,100 +62,59 @@ export class GraphQLServer {
6262
}
6363

6464
/**
65-
* Fetches the list of active platforms from the registry
66-
* @returns Promise<string[]> - Array of platform URLs
65+
* Forwards an awareness packet to Awareness as a Service (AaaS).
66+
*
67+
* AaaS has replaced eVault's built-in webhook fanout: instead of querying
68+
* the registry and POSTing to every platform here, we make a single POST
69+
* to AaaS, which owns subscription matching, retry/dead-letter delivery and
70+
* the catch-all fanout that preserves the previous behaviour.
71+
*
72+
* @param webhookPayload - The awareness packet { id, w3id, evaultPublicKey,
73+
* data, schemaId }
74+
* @param requestingPlatform - The platform that triggered the change, if
75+
* known. AaaS uses it to skip delivering the packet
76+
* back to its origin (prevents webhook ping-pong).
6777
*/
68-
private async getActivePlatforms(): Promise<string[]> {
69-
try {
70-
if (!process.env.PUBLIC_REGISTRY_URL) {
71-
return [];
72-
}
73-
74-
const response = await axios.get(
75-
new URL(
76-
"/platforms",
77-
process.env.PUBLIC_REGISTRY_URL,
78-
).toString(),
79-
);
80-
return response.data;
81-
} catch (error) {
82-
return [];
83-
}
84-
}
85-
86-
/**
87-
* Delivers webhooks to all platforms except the requesting one
88-
* @param requestingPlatform - The platform that made the request (if any)
89-
* @param webhookPayload - The payload to send to webhooks
90-
*/
91-
private async deliverWebhooks(
92-
requestingPlatform: string | null,
78+
private async notifyAwareness(
9379
webhookPayload: any,
80+
requestingPlatform: string | null = null,
9481
): Promise<void> {
95-
// One log line per dispatch — the same payload goes to every
96-
// target platform, so we log the body once here instead of per
97-
// target. This is the source of truth for "what eVault claims it
98-
// sent"; correlate against receiver logs to find divergence.
82+
// One log line per dispatch — this remains the source of truth for
83+
// "what eVault claims it sent"; correlate against AaaS ingest logs.
9984
try {
10085
const payloadJson = JSON.stringify(webhookPayload);
10186
console.log(
102-
`[webhook] id=${webhookPayload?.id} schemaId=${webhookPayload?.schemaId} w3id=${webhookPayload?.w3id} from=${requestingPlatform ?? "<none>"} payload=${payloadJson}`,
87+
`[webhook] id=${webhookPayload?.id} schemaId=${webhookPayload?.schemaId} w3id=${webhookPayload?.w3id} payload=${payloadJson}`,
10388
);
10489
} catch {
10590
console.log(
10691
`[webhook] id=${webhookPayload?.id} schemaId=${webhookPayload?.schemaId} payload=<unserializable>`,
10792
);
10893
}
10994

110-
try {
111-
const activePlatforms = await this.getActivePlatforms();
112-
113-
// Filter out the requesting platform
114-
const platformsToNotify = activePlatforms.filter((platformUrl) => {
115-
if (!requestingPlatform) return true;
116-
117-
try {
118-
// Normalize URLs for comparison
119-
const normalizedPlatformUrl = new URL(
120-
platformUrl,
121-
).toString();
122-
const normalizedRequestingPlatform = new URL(
123-
requestingPlatform,
124-
).toString();
125-
126-
return (
127-
normalizedPlatformUrl !== normalizedRequestingPlatform
128-
);
129-
} catch (error) {
130-
// If requestingPlatform is not a valid URL, don't filter it out
131-
// (treat it as a different platform identifier)
132-
return true;
133-
}
134-
});
95+
if (!process.env.AWARENESS_SERVICE_URL) {
96+
console.log("[webhook] AWARENESS_SERVICE_URL not set, skipping");
97+
return;
98+
}
13599

136-
// Send webhooks to all other platforms
137-
const webhookPromises = platformsToNotify.map(
138-
async (platformUrl) => {
139-
try {
140-
const webhookUrl = new URL(
141-
"/api/webhook",
142-
platformUrl,
143-
).toString();
144-
await axios.post(webhookUrl, webhookPayload, {
145-
headers: {
146-
"Content-Type": "application/json",
147-
},
148-
timeout: 5000, // 5 second timeout
149-
});
150-
} catch (error) {
151-
console.log(`Webhook delivery failed to ${platformUrl}`);
152-
}
100+
try {
101+
await axios.post(
102+
new URL(
103+
"/ingest",
104+
process.env.AWARENESS_SERVICE_URL,
105+
).toString(),
106+
{ ...webhookPayload, requestingPlatform },
107+
{
108+
headers: {
109+
"Content-Type": "application/json",
110+
"x-ingest-secret":
111+
process.env.AWARENESS_INGEST_SECRET ?? "",
112+
},
113+
timeout: 5000,
153114
},
154115
);
155-
156-
await Promise.allSettled(webhookPromises);
157116
} catch (error) {
158-
console.log("Webhook delivery failed");
117+
console.log("Awareness ingest delivery failed");
159118
}
160119
}
161120

@@ -404,9 +363,7 @@ export class GraphQLServer {
404363
parsed: parsedFromEnvelopes,
405364
};
406365

407-
// Deliver webhooks for create operation
408-
const requestingPlatform =
409-
context.tokenPayload?.platform || null;
366+
// Forward the awareness packet for create operation
410367
const webhookPayload = {
411368
id: result.metaEnvelope.id,
412369
w3id: context.eName,
@@ -415,13 +372,11 @@ export class GraphQLServer {
415372
schemaId: input.ontology,
416373
};
417374

418-
// Delayed webhook delivery to prevent ping-pong
419-
setTimeout(() => {
420-
this.deliverWebhooks(
421-
requestingPlatform,
422-
webhookPayload,
423-
);
424-
}, 3_000);
375+
// Fire-and-forget ingest to AaaS
376+
this.notifyAwareness(
377+
webhookPayload,
378+
context.tokenPayload?.platform || null,
379+
);
425380

426381
// Send push notifications for new messages
427382
console.log(`[NOTIF] createMetaEnvelope ontology: "${input.ontology}"`);
@@ -553,8 +508,6 @@ export class GraphQLServer {
553508
// would make the receiver lose every untouched
554509
// field (e.g. a read-receipt update would wipe
555510
// participantIds on the receiver side).
556-
const requestingPlatform =
557-
context.tokenPayload?.platform || null;
558511
const webhookPayload = {
559512
id,
560513
w3id: context.eName,
@@ -563,10 +516,10 @@ export class GraphQLServer {
563516
schemaId: input.ontology,
564517
};
565518

566-
// Fire and forget webhook delivery
567-
this.deliverWebhooks(
568-
requestingPlatform,
519+
// Fire-and-forget ingest to AaaS
520+
this.notifyAwareness(
569521
webhookPayload,
522+
context.tokenPayload?.platform || null,
570523
);
571524

572525
// Log envelope operation best-effort
@@ -777,10 +730,8 @@ export class GraphQLServer {
777730
});
778731
successCount++;
779732

780-
// Deliver webhooks if not skipping
733+
// Forward awareness packet if not skipping
781734
if (!shouldSkipWebhooks) {
782-
const requestingPlatform =
783-
context.tokenPayload?.platform || null;
784735
const webhookPayload = {
785736
id: result.metaEnvelope.id,
786737
w3id: context.eName,
@@ -789,12 +740,12 @@ export class GraphQLServer {
789740
schemaId: input.ontology,
790741
};
791742

792-
// Fire and forget webhook delivery
793-
this.deliverWebhooks(
794-
requestingPlatform,
743+
// Fire-and-forget ingest to AaaS
744+
this.notifyAwareness(
795745
webhookPayload,
746+
context.tokenPayload?.platform || null,
796747
).catch((err) => {
797-
console.error(`[WEBHOOK] Delivery failed for bulk-create envelope ${result.metaEnvelope.id}:`, err);
748+
console.error(`[WEBHOOK] AaaS ingest failed for bulk-create envelope ${result.metaEnvelope.id}:`, err);
798749
});
799750
}
800751

@@ -949,8 +900,6 @@ export class GraphQLServer {
949900
),
950901
);
951902

952-
const requestingPlatform =
953-
context.tokenPayload?.platform || null;
954903
const webhookPayload = {
955904
id: metaEnvelopeId,
956905
w3id: context.eName,
@@ -959,12 +908,10 @@ export class GraphQLServer {
959908
schemaId:
960909
BINDING_DOCUMENT_ONTOLOGY,
961910
};
962-
setTimeout(() => {
963-
this.deliverWebhooks(
964-
requestingPlatform,
965-
webhookPayload,
966-
);
967-
}, 3_000);
911+
this.notifyAwareness(
912+
webhookPayload,
913+
context.tokenPayload?.platform || null,
914+
);
968915

969916
return {
970917
bindingDocument: result.bindingDocument,
@@ -1059,8 +1006,6 @@ export class GraphQLServer {
10591006
),
10601007
);
10611008

1062-
const requestingPlatform =
1063-
context.tokenPayload?.platform || null;
10641009
const webhookPayload = {
10651010
id: input.bindingDocumentId,
10661011
w3id: context.eName,
@@ -1069,12 +1014,10 @@ export class GraphQLServer {
10691014
schemaId:
10701015
BINDING_DOCUMENT_ONTOLOGY,
10711016
};
1072-
setTimeout(() => {
1073-
this.deliverWebhooks(
1074-
requestingPlatform,
1075-
webhookPayload,
1076-
);
1077-
}, 3_000);
1017+
this.notifyAwareness(
1018+
webhookPayload,
1019+
context.tokenPayload?.platform || null,
1020+
);
10781021

10791022
return {
10801023
bindingDocument: result,
@@ -1138,9 +1081,10 @@ export class GraphQLServer {
11381081
parsed: input.payload,
11391082
};
11401083

1141-
// Deliver webhooks for create operation
1142-
const requestingPlatform =
1143-
context.tokenPayload?.platform || null;
1084+
// Forward the awareness packet for create operation.
1085+
// The requesting platform is passed so AaaS can skip
1086+
// delivering the packet back to its origin — the same
1087+
// ping-pong guard the old fanout enforced here.
11441088
const webhookPayload = {
11451089
id: result.metaEnvelope.id,
11461090
w3id: context.eName,
@@ -1149,22 +1093,10 @@ export class GraphQLServer {
11491093
schemaId: input.ontology,
11501094
};
11511095

1152-
/**
1153-
* To whoever who reads this in the future please don't
1154-
* remove this delay as this prevents a VERY horrible
1155-
* disgusting edge case, where if a platform's URL is
1156-
* not determinable the webhook to the same platform as
1157-
* the one who sent off the request gets sent and that
1158-
* is not an ideal case trust me I've suffered, it
1159-
* causes an absolutely beautiful error where you get
1160-
* stuck in what I like to call webhook ping-pong
1161-
*/
1162-
setTimeout(() => {
1163-
this.deliverWebhooks(
1164-
requestingPlatform,
1165-
webhookPayload,
1166-
);
1167-
}, 3_000);
1096+
this.notifyAwareness(
1097+
webhookPayload,
1098+
context.tokenPayload?.platform || null,
1099+
);
11681100

11691101
// Send push notifications for new messages
11701102
console.log(`[NOTIF] storeMetaEnvelope ontology: "${input.ontology}"`);
@@ -1249,8 +1181,6 @@ export class GraphQLServer {
12491181
// resolver above — sending input.payload (the
12501182
// partial diff) would make receivers clobber their
12511183
// own untouched fields.
1252-
const requestingPlatform =
1253-
context.tokenPayload?.platform || null;
12541184
const webhookPayload = {
12551185
id: id,
12561186
w3id: context.eName,
@@ -1259,10 +1189,10 @@ export class GraphQLServer {
12591189
schemaId: input.ontology,
12601190
};
12611191

1262-
// Fire and forget webhook delivery
1263-
this.deliverWebhooks(
1264-
requestingPlatform,
1192+
// Fire-and-forget ingest to AaaS
1193+
this.notifyAwareness(
12651194
webhookPayload,
1195+
context.tokenPayload?.platform || null,
12661196
);
12671197

12681198
// Log envelope operation best-effort (do not fail mutation)

‎services/awareness-service/api/src/services/IngestService.ts‎

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,15 @@ import { Packet } from "../database/entities/Packet";
44
import type { AwarenessPayload } from "../types";
55
import { SubscriptionMatcher } from "./SubscriptionMatcher";
66

7+
/** Returns the normalised origin of a URL, or null if it cannot be parsed. */
8+
function safeOrigin(url: string): string | null {
9+
try {
10+
return new URL(url).origin;
11+
} catch {
12+
return null;
13+
}
14+
}
15+
716
/**
817
* Persists an incoming awareness packet and queues a webhook delivery for every
918
* subscription that matches it. Re-ingesting the same packet is idempotent: the
@@ -30,7 +39,19 @@ export class IngestService {
3039

3140
await packetRepo.upsert(packet, ["id"]);
3241

33-
const subscriptions = await this.matcher.match(packet);
42+
let subscriptions = await this.matcher.match(packet);
43+
44+
// Skip delivering the packet back to the platform that triggered it -
45+
// the same ping-pong guard evault-core's old fanout enforced.
46+
if (payload.requestingPlatform) {
47+
const origin = safeOrigin(payload.requestingPlatform);
48+
if (origin) {
49+
subscriptions = subscriptions.filter(
50+
(sub) => safeOrigin(sub.targetUrl) !== origin,
51+
);
52+
}
53+
}
54+
3455
if (subscriptions.length === 0) {
3556
return { packetId: packet.id, deliveriesQueued: 0 };
3657
}

‎services/awareness-service/api/src/types.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,11 @@ export interface AwarenessPayload {
88
data?: Record<string, unknown> | null;
99
schemaId: string;
1010
operation?: "create" | "update" | "delete";
11+
/**
12+
* The platform that triggered the change, if known. Used only to skip
13+
* delivering the packet back to its origin; never persisted or delivered.
14+
*/
15+
requestingPlatform?: string | null;
1116
}
1217

1318
declare global {

0 commit comments

Comments
 (0)