Prefer local telegram forward after restore

johnlindquist · johnlindquist · commit a0fbbdd1a6c5 · 2026-04-13T16:09:35.000-06:00
diff --git a/src/server/workflows/channels/drain-channel-workflow.test.ts b/src/server/workflows/channels/drain-channel-workflow.test.ts
@@ -61,12 +61,29 @@ function createWorkflowDependencies(
         channels: { telegram: null, slack: null, discord: null, whatsapp: null },
       }),
     getSandboxDomain: async () => "https://sandbox.example.test",
-    forwardToNativeHandler: async () => ({ ok: true, status: 200 }),
+    forwardToNativeHandler: async () => ({
+      ok: true,
+      status: 200,
+      durationMs: 0,
+      bodyLength: 0,
+      bodyHead: "",
+      headers: null,
+    }),
+    forwardTelegramToNativeHandlerLocally: async () => ({
+      ok: true,
+      status: 200,
+      durationMs: 0,
+      bodyLength: 0,
+      bodyHead: "",
+      headers: null,
+      error: null,
+    }),
     forwardToNativeHandlerWithRetry: async (): Promise<RetryingForwardResult> => ({
       ok: true,
       status: 200,
       attempts: 1,
       totalMs: 50,
+      transport: "public",
       retries: [],
     }),
     waitForTelegramNativeHandler: async (): Promise<TelegramProbeResult> => ({
@@ -108,7 +125,7 @@ test("processChannelStep skips ensureSandboxReady when boot returns running", as
     },
     forwardToNativeHandlerWithRetry: async (_channel: unknown, _payload: unknown, meta: SingleMeta): Promise<RetryingForwardResult> => {
       forwardedSandboxId = meta.sandboxId ?? null;
-      return { ok: true, status: 200, attempts: 1, totalMs: 50, retries: [] };
+      return { ok: true, status: 200, attempts: 1, totalMs: 50, transport: "public", retries: [] };
     },
   });
 
@@ -139,7 +156,7 @@ test("processChannelStep falls back to ensureSandboxReady when boot returns non-
     },
     forwardToNativeHandlerWithRetry: async (_channel: unknown, _payload: unknown, meta: SingleMeta): Promise<RetryingForwardResult> => {
       forwardedSandboxId = meta.sandboxId ?? null;
-      return { ok: true, status: 200, attempts: 1, totalMs: 50, retries: [] };
+      return { ok: true, status: 200, attempts: 1, totalMs: 50, transport: "public", retries: [] };
     },
   });
 
@@ -173,6 +190,7 @@ test("processChannelStep converts native forward 502 into RetryableError (Telegr
       status: 502,
       attempts: 6,
       totalMs: 6000,
+      transport: "public",
       retries: [{ attempt: 1, reason: "proxy-error", status: 502 }],
     }),
   });
@@ -194,6 +212,7 @@ test("processChannelStep keeps native forward 404 fatal (Telegram retrying path)
       status: 404,
       attempts: 1,
       totalMs: 50,
+      transport: "public",
       retries: [],
     }),
   });
@@ -214,11 +233,11 @@ test("processChannelStep uses retrying forward for Telegram, direct forward for
   const telegramDeps = createWorkflowDependencies({
     forwardToNativeHandlerWithRetry: async (): Promise<RetryingForwardResult> => {
       retryingCalled = true;
-      return { ok: true, status: 200, attempts: 1, totalMs: 50, retries: [] };
+      return { ok: true, status: 200, attempts: 1, totalMs: 50, transport: "public", retries: [] };
     },
     forwardToNativeHandler: async () => {
       directCalled = true;
-      return { ok: true, status: 200 };
+      return { ok: true, status: 200, durationMs: 0, bodyLength: 0, bodyHead: "", headers: null };
     },
   });
 
@@ -232,11 +251,11 @@ test("processChannelStep uses retrying forward for Telegram, direct forward for
   const slackDeps = createWorkflowDependencies({
     forwardToNativeHandlerWithRetry: async (): Promise<RetryingForwardResult> => {
       retryingCalled = true;
-      return { ok: true, status: 200, attempts: 1, totalMs: 50, retries: [] };
+      return { ok: true, status: 200, attempts: 1, totalMs: 50, transport: "public", retries: [] };
     },
     forwardToNativeHandler: async () => {
       directCalled = true;
-      return { ok: true, status: 200 };
+      return { ok: true, status: 200, durationMs: 0, bodyLength: 0, bodyHead: "", headers: null };
     },
   });
 
@@ -252,6 +271,7 @@ test("processChannelStep converts retrying forward 504 (exhausted) into Retryabl
       status: 504,
       attempts: 6,
       totalMs: 30000,
+      transport: null,
       retries: [
         { attempt: 1, reason: "proxy-error", status: 503 },
         { attempt: 2, reason: "fetch-exception", error: "connect ECONNREFUSED" },
@@ -279,6 +299,7 @@ test("processChannelStep treats retrying forward 500 as retryable at workflow le
       status: 500,
       attempts: 1,
       totalMs: 100,
+      transport: "public",
       retries: [],
     }),
   });
@@ -515,7 +536,7 @@ test("processChannelStep forward captures Telegram webhook secret and correct po
       capturedPayload = payload;
       capturedMeta = meta;
       capturedGetSandboxDomain = getSandboxDomain;
-      return { ok: true, status: 200, attempts: 1, totalMs: 50, retries: [] };
+      return { ok: true, status: 200, attempts: 1, totalMs: 50, transport: "public", retries: [] };
     },
   });
 
@@ -554,7 +575,7 @@ test("processChannelStep forward passes meta with portUrls from boot result", as
       meta: SingleMeta,
     ): Promise<RetryingForwardResult> => {
       forwardedPortUrls = (meta.portUrls as Record<string, string>) ?? null;
-      return { ok: true, status: 200, attempts: 1, totalMs: 50, retries: [] };
+      return { ok: true, status: 200, attempts: 1, totalMs: 50, transport: "public", retries: [] };
     },
   });
 
@@ -610,7 +631,7 @@ test("processChannelStep uses local Telegram native handler readiness before for
     },
     forwardToNativeHandlerWithRetry: async (): Promise<RetryingForwardResult> => {
       forwardCalledAfterLocalProbe = localProbeCallCount > 0;
-      return { ok: true, status: 200, attempts: 1, totalMs: 50, retries: [] };
+      return { ok: true, status: 200, attempts: 1, totalMs: 50, transport: "local", retries: [] };
     },
   });
 
@@ -680,7 +701,7 @@ test("processChannelStep falls back to public Telegram probe when local handler
     },
     forwardToNativeHandlerWithRetry: async (): Promise<RetryingForwardResult> => {
       forwardCalled = true;
-      return { ok: true, status: 200, attempts: 1, totalMs: 50, retries: [] };
+      return { ok: true, status: 200, attempts: 1, totalMs: 50, transport: "public", retries: [] };
     },
   });
 
@@ -726,7 +747,7 @@ test("processChannelStep still forwards when both Telegram probes time out", asy
     },
     forwardToNativeHandlerWithRetry: async (): Promise<RetryingForwardResult> => {
       forwardCalled = true;
-      return { ok: true, status: 200, attempts: 1, totalMs: 50, retries: [] };
+      return { ok: true, status: 200, attempts: 1, totalMs: 50, transport: "public", retries: [] };
     },
   });
 
diff --git a/src/server/workflows/channels/drain-channel-workflow.ts b/src/server/workflows/channels/drain-channel-workflow.ts
@@ -17,6 +17,7 @@ export type RetryingForwardResult = {
   status: number;
   attempts: number;
   totalMs: number;
+  transport?: "public" | "local" | null;
   retries: Array<{ attempt: number; reason: string; status?: number; error?: string }>;
   attemptsDetail?: ForwardAttemptDetail[];
 };
@@ -287,7 +288,8 @@ export async function processChannelStep(
         payload,
         readyMeta,
         getSandboxDomain,
-        localProbeResult?.ready === true ? null : forwardTelegramToNativeHandlerLocally,
+        forwardTelegramToNativeHandlerLocally,
+        localProbeResult?.ready === true,
       );
       forwardResult = { ok: retryingResult.ok, status: retryingResult.status };
     } else {
@@ -306,6 +308,8 @@ export async function processChannelStep(
     diag.forwardAttempts = retryingResult?.attempts ?? null;
     diag.forwardRetries = retryingResult?.retries ?? null;
     diag.forwardTotalMs = retryingResult?.totalMs ?? null;
+    diag.forwardTransport =
+      retryingResult?.transport ?? (channel === "telegram" ? "public" : null);
     diag.forwardAttemptTimeline = retryingResult?.attemptsDetail ?? null;
     console.log(`[DIAG] Phase 2 DONE: ok=${forwardResult.ok} status=${forwardResult.status} attempts=${retryingResult?.attempts ?? 1} retries=${JSON.stringify(retryingResult?.retries ?? [])} durationMs=${diag.forwardDurationMs}`);
     await persistDiagSnapshot("native-forward-complete", {
@@ -315,6 +319,7 @@ export async function processChannelStep(
       forwardAttempts: diag.forwardAttempts,
       forwardRetries: diag.forwardRetries,
       forwardTotalMs: diag.forwardTotalMs,
+      forwardTransport: diag.forwardTransport,
       forwardAttemptTimeline: diag.forwardAttemptTimeline,
     });
 
@@ -324,6 +329,7 @@ export async function processChannelStep(
       sandboxId: readyMeta.sandboxId,
       ok: forwardResult.ok,
       status: forwardResult.status,
+      transport: retryingResult?.transport ?? (channel === "telegram" ? "public" : null),
       retryingForwardAttempts: retryingResult?.attempts ?? null,
       retryingForwardTotalMs: retryingResult?.totalMs ?? null,
       retryingForwardRetries: retryingResult?.retries?.length ?? null,
@@ -368,6 +374,7 @@ export async function processChannelStep(
         telegramLocalProbeHeaders: diag.telegramLocalProbeHeaders ?? null,
         retryingForwardAttempts: retryingResult?.attempts ?? null,
         retryingForwardTotalMs: retryingResult?.totalMs ?? null,
+        retryingForwardTransport: retryingResult?.transport ?? null,
         retryingForwardAttemptTimeline: retryingResult?.attemptsDetail ?? null,
         telegramReconcileBlocking: restore?.telegramReconcileBlocking ?? null,
         telegramReconcileMs: restore?.telegramReconcileMs ?? null,
@@ -497,6 +504,7 @@ export type ForwardAttemptDetail = {
   bodyLength: number | null;
   bodyHead: string | null;
   headers: DiagnosticHeaders | null;
+  transport: "public" | "local";
   classification: string;
   error?: string | null;
 };
@@ -928,6 +936,7 @@ async function forwardToNativeHandlerWithRetry(
     headers: DiagnosticHeaders | null;
     error?: string | null;
   }>) | null,
+  preferLocalTelegramForward = false,
 ): Promise<RetryingForwardResult> {
   const startedAt = Date.now();
   const deadline = startedAt + RETRYING_FORWARD_TIMEOUT_MS;
@@ -937,9 +946,15 @@ async function forwardToNativeHandlerWithRetry(
   for (let attempt = 1; attempt <= RETRYING_FORWARD_MAX_ATTEMPTS && Date.now() < deadline; attempt++) {
     const attemptStartedAt = Date.now();
     try {
-      const result = channel === "telegram" && meta.sandboxId && forwardTelegramToNativeHandlerLocally
+      const useLocalForward =
+        channel === "telegram"
+        && meta.sandboxId != null
+        && forwardTelegramToNativeHandlerLocally != null
+        && preferLocalTelegramForward;
+      const transport: "public" | "local" = useLocalForward ? "local" : "public";
+      const result = useLocalForward
         ? await forwardTelegramToNativeHandlerLocally(
-            meta.sandboxId,
+            meta.sandboxId as string,
             payload,
             meta.channels.telegram?.webhookSecret ?? null,
           )
@@ -982,6 +997,7 @@ async function forwardToNativeHandlerWithRetry(
         bodyLength: result.bodyLength,
         bodyHead: result.bodyHead,
         headers: result.headers,
+        transport,
         classification,
       });
       if (result.status >= 502 || result.status === 401 || result.status === 404 || swallowed) {
@@ -996,6 +1012,7 @@ async function forwardToNativeHandlerWithRetry(
           durationMs: result.durationMs,
           bodyLength: result.bodyLength,
           bodyHead: result.bodyHead,
+          transport,
           responseHeaders: result.headers,
           retryElapsedMs: Date.now() - startedAt,
         });
@@ -1014,6 +1031,7 @@ async function forwardToNativeHandlerWithRetry(
         status: result.status,
         attempts: attempt,
         totalMs,
+        transport,
         retryCount: retries.length,
         attemptsDetail,
       });
@@ -1022,6 +1040,7 @@ async function forwardToNativeHandlerWithRetry(
         status: result.status,
         attempts: attempt,
         totalMs,
+        transport,
         retries,
         attemptsDetail,
       };
@@ -1040,6 +1059,7 @@ async function forwardToNativeHandlerWithRetry(
         bodyLength: null,
         bodyHead: null,
         headers: null,
+        transport: "public",
         classification: "fetch-exception",
         error: errorMsg,
       });
@@ -1070,6 +1090,7 @@ async function forwardToNativeHandlerWithRetry(
     status: 504,
     attempts: RETRYING_FORWARD_MAX_ATTEMPTS,
     totalMs,
+    transport: null,
     retries,
     attemptsDetail,
   };
