Skip to content

Commit 12cf34a

Browse files
vincentkocvincentkoc
committed
refactor: share send inflight helpers
1 parent d328a0d commit 12cf34a

1 file changed

Lines changed: 100 additions & 32 deletions

File tree

‎src/gateway/server-methods/send.ts‎

Lines changed: 100 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -95,22 +95,27 @@ function resolveGatewayInflightMap(params: { context: GatewayRequestContext; ded
9595
return { kind: "ready", inflightMap };
9696
}
9797

98-
function resolveGatewayInflightStart(params: {
98+
function resolveGatewayInflightRequest(params: {
9999
context: GatewayRequestContext;
100-
dedupeKey: string;
100+
prefix: "message.action" | "poll" | "send";
101+
idempotencyKey: string;
101102
respond: RespondFn;
102103
}):
103104
| {
104105
kind: "ready";
106+
idem: string;
107+
dedupeKey: string;
105108
inflightMap: Map<string, Promise<InflightResult>>;
106109
}
107110
| {
108111
kind: "handled";
109112
done: Promise<void>;
110113
} {
114+
const idem = params.idempotencyKey;
115+
const dedupeKey = `${params.prefix}:${idem}`;
111116
const inflight = resolveGatewayInflightMap({
112117
context: params.context,
113-
dedupeKey: params.dedupeKey,
118+
dedupeKey,
114119
});
115120
if (inflight.kind === "cached") {
116121
params.respond(inflight.cached.ok, inflight.cached.payload, inflight.cached.error, {
@@ -127,7 +132,12 @@ function resolveGatewayInflightStart(params: {
127132
}),
128133
};
129134
}
130-
return { kind: "ready", inflightMap: inflight.inflightMap };
135+
return {
136+
kind: "ready",
137+
idem,
138+
dedupeKey,
139+
inflightMap: inflight.inflightMap,
140+
};
131141
}
132142

133143
async function runGatewayInflightWork(params: {
@@ -191,6 +201,36 @@ async function resolveRequestedChannel(params: {
191201
return { cfg, sourceCfg, channel };
192202
}
193203

204+
async function resolveInternalDeliveryChannel(
205+
requestChannel: unknown,
206+
context: GatewayRequestContext,
207+
): Promise<
208+
| {
209+
kind: "ready";
210+
cfg: OpenClawConfig;
211+
sourceCfg: OpenClawConfig;
212+
channel: string;
213+
}
214+
| {
215+
kind: "failed";
216+
result: InflightResult;
217+
}
218+
> {
219+
const resolvedChannel = await resolveRequestedChannel({
220+
requestChannel,
221+
unsupportedMessage: (input) => `unsupported channel: ${input}`,
222+
context,
223+
rejectWebchatAsInternalOnly: true,
224+
});
225+
if ("error" in resolvedChannel) {
226+
return {
227+
kind: "failed",
228+
result: { ok: false, error: resolvedChannel.error },
229+
};
230+
}
231+
return { kind: "ready", ...resolvedChannel };
232+
}
233+
194234
function resolveGatewayOutboundTarget(params: {
195235
channel: string;
196236
to: string;
@@ -312,6 +352,25 @@ function createGatewayInflightSuccess(params: {
312352
};
313353
}
314354

355+
function createGatewayDeliveryInflightSuccess(params: {
356+
context: GatewayRequestContext;
357+
dedupeKey: string;
358+
runId: string;
359+
channel: string;
360+
result: Record<string, unknown>;
361+
}): InflightResult {
362+
return createGatewayInflightSuccess({
363+
context: params.context,
364+
dedupeKey: params.dedupeKey,
365+
payload: buildGatewayDeliveryPayload({
366+
runId: params.runId,
367+
channel: params.channel,
368+
result: params.result,
369+
}),
370+
channel: params.channel,
371+
});
372+
}
373+
315374
function createGatewayInflightUnavailableFailure(params: {
316375
context: GatewayRequestContext;
317376
dedupeKey: string;
@@ -411,14 +470,17 @@ export const sendHandlers: GatewayRequestHandlers = {
411470
};
412471
idempotencyKey: string;
413472
};
414-
const idem = request.idempotencyKey;
415-
const dedupeKey = `message.action:${idem}`;
416-
const inflightStart = resolveGatewayInflightStart({ context, dedupeKey, respond });
417-
if (inflightStart.kind === "handled") {
418-
await inflightStart.done;
473+
const inflight = resolveGatewayInflightRequest({
474+
context,
475+
prefix: "message.action",
476+
idempotencyKey: request.idempotencyKey,
477+
respond,
478+
});
479+
if (inflight.kind === "handled") {
480+
await inflight.done;
419481
return;
420482
}
421-
const inflightMap = inflightStart.inflightMap;
483+
const { dedupeKey, inflightMap } = inflight;
422484
const work = (async (): Promise<InflightResult> => {
423485
const resolvedChannel = await resolveRequestedChannel({
424486
requestChannel: request.channel,
@@ -532,14 +594,17 @@ export const sendHandlers: GatewayRequestHandlers = {
532594
sessionKey?: string;
533595
idempotencyKey: string;
534596
};
535-
const idem = request.idempotencyKey;
536-
const dedupeKey = `send:${idem}`;
537-
const inflightStart = resolveGatewayInflightStart({ context, dedupeKey, respond });
538-
if (inflightStart.kind === "handled") {
539-
await inflightStart.done;
597+
const inflight = resolveGatewayInflightRequest({
598+
context,
599+
prefix: "send",
600+
idempotencyKey: request.idempotencyKey,
601+
respond,
602+
});
603+
if (inflight.kind === "handled") {
604+
await inflight.done;
540605
return;
541606
}
542-
const inflightMap = inflightStart.inflightMap;
607+
const { idem, dedupeKey, inflightMap } = inflight;
543608
const to = normalizeOptionalString(request.to) ?? "";
544609
const message = normalizeOptionalString(request.message) ?? "";
545610
const mediaUrl = normalizeOptionalString(request.mediaUrl);
@@ -561,14 +626,9 @@ export const sendHandlers: GatewayRequestHandlers = {
561626
const threadId = normalizeOptionalString(request.threadId);
562627

563628
const work = (async (): Promise<InflightResult> => {
564-
const resolvedChannel = await resolveRequestedChannel({
565-
requestChannel: request.channel,
566-
unsupportedMessage: (input) => `unsupported channel: ${input}`,
567-
context,
568-
rejectWebchatAsInternalOnly: true,
569-
});
570-
if ("error" in resolvedChannel) {
571-
return { ok: false, error: resolvedChannel.error };
629+
const resolvedChannel = await resolveInternalDeliveryChannel(request.channel, context);
630+
if (resolvedChannel.kind !== "ready") {
631+
return resolvedChannel.result;
572632
}
573633
const { cfg, channel } = resolvedChannel;
574634
const outboundChannel = channel;
@@ -709,8 +769,13 @@ export const sendHandlers: GatewayRequestHandlers = {
709769
if (!result) {
710770
throw new Error("No delivery result");
711771
}
712-
const payload = buildGatewayDeliveryPayload({ runId: idem, channel, result });
713-
return createGatewayInflightSuccess({ context, dedupeKey, payload, channel });
772+
return createGatewayDeliveryInflightSuccess({
773+
context,
774+
dedupeKey,
775+
runId: idem,
776+
channel,
777+
result,
778+
});
714779
} catch (err) {
715780
return createGatewayInflightUnavailableFailure({ context, dedupeKey, channel, err });
716781
}
@@ -745,14 +810,17 @@ export const sendHandlers: GatewayRequestHandlers = {
745810
accountId?: string;
746811
idempotencyKey: string;
747812
};
748-
const idem = request.idempotencyKey;
749-
const dedupeKey = `poll:${idem}`;
750-
const inflightStart = resolveGatewayInflightStart({ context, dedupeKey, respond });
751-
if (inflightStart.kind === "handled") {
752-
await inflightStart.done;
813+
const inflight = resolveGatewayInflightRequest({
814+
context,
815+
prefix: "poll",
816+
idempotencyKey: request.idempotencyKey,
817+
respond,
818+
});
819+
if (inflight.kind === "handled") {
820+
await inflight.done;
753821
return;
754822
}
755-
const inflightMap = inflightStart.inflightMap;
823+
const { idem, dedupeKey, inflightMap } = inflight;
756824
const work = (async (): Promise<InflightResult> => {
757825
const resolvedChannel = await resolveRequestedChannel({
758826
requestChannel: request.channel,

0 commit comments

Comments
 (0)