diff --git a/extensions/line/src/download.test.ts b/extensions/line/src/download.test.ts index 44c497eca6dd..c4c317534607 100644 --- a/extensions/line/src/download.test.ts +++ b/extensions/line/src/download.test.ts @@ -1,19 +1,27 @@ // Line tests cover download plugin behavior. -import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -const getMessageContentMock = vi.hoisted(() => vi.fn()); +const fetchMock = vi.hoisted(() => vi.fn()); +const delayMock = vi.hoisted(() => vi.fn()); const saveMediaStreamMock = vi.hoisted(() => vi.fn()); -vi.mock("@line/bot-sdk", () => ({ - messagingApi: { - MessagingApiBlobClient: class { - getMessageContent(messageId: string) { - return getMessageContentMock(messageId); - } - }, - }, +vi.mock("node:timers/promises", () => ({ + setTimeout: delayMock, })); +function responseWithChunks(status: number, parts: Buffer[]): Response { + return new Response(Buffer.concat(parts), { status }); +} + +function cancellableResponse(status: number): { + response: Response; + cancel: ReturnType; +} { + const cancel = vi.fn(); + const body = new ReadableStream({ cancel }); + return { response: new Response(body, { status }), cancel }; +} + vi.mock("openclaw/plugin-sdk/runtime-env", () => ({ createSubsystemLogger: () => { const logger = { @@ -34,12 +42,6 @@ vi.mock("openclaw/plugin-sdk/media-store", () => ({ let downloadLineMedia: typeof import("./download.js").downloadLineMedia; -async function* chunks(parts: Buffer[]): AsyncGenerator { - for (const part of parts) { - yield part; - } -} - function saveMediaStreamCall(): unknown[] { const call = saveMediaStreamMock.mock.calls.at(0); if (!call) { @@ -64,15 +66,22 @@ describe("downloadLineMedia", () => { }); afterAll(() => { - vi.doUnmock("@line/bot-sdk"); + vi.doUnmock("node:timers/promises"); vi.doUnmock("openclaw/plugin-sdk/runtime-env"); vi.doUnmock("openclaw/plugin-sdk/media-store"); + vi.unstubAllGlobals(); vi.resetModules(); }); + afterEach(() => { + vi.useRealTimers(); + }); + beforeEach(() => { vi.restoreAllMocks(); - getMessageContentMock.mockReset(); + vi.stubGlobal("fetch", fetchMock); + fetchMock.mockReset(); + delayMock.mockReset().mockResolvedValue(undefined); saveMediaStreamMock.mockReset(); saveMediaStreamMock.mockImplementation( async (stream: AsyncIterable, contentType?: string, subdir?: string) => { @@ -92,10 +101,18 @@ describe("downloadLineMedia", () => { it("persists inbound media with the shared media store", async () => { const jpeg = Buffer.from([0xff, 0xd8, 0xff, 0x00]); - getMessageContentMock.mockResolvedValueOnce(chunks([jpeg])); + fetchMock.mockResolvedValueOnce(responseWithChunks(200, [jpeg])); const result = await downloadLineMedia("mid-jpeg", "token"); + expect(fetchMock).toHaveBeenCalledWith( + "https://api-data.line.me/v2/bot/message/mid-jpeg/content", + expect.objectContaining({ + headers: { Authorization: "Bearer token" }, + redirect: "error", + signal: expect.any(AbortSignal), + }), + ); expect(saveMediaStreamMock).toHaveBeenCalledTimes(1); const call = saveMediaStreamCall(); expect(call[1]).toBeUndefined(); @@ -111,10 +128,13 @@ describe("downloadLineMedia", () => { it("does not pass the external messageId to saveMediaStream", async () => { const messageId = "a/../../../../etc/passwd"; const jpeg = Buffer.from([0xff, 0xd8, 0xff, 0x00]); - getMessageContentMock.mockResolvedValueOnce(chunks([jpeg])); + fetchMock.mockResolvedValueOnce(responseWithChunks(200, [jpeg])); const result = await downloadLineMedia(messageId, "token"); + expect(fetchMock.mock.calls[0]?.[0]).toBe( + "https://api-data.line.me/v2/bot/message/a%2F..%2F..%2F..%2F..%2Fetc%2Fpasswd/content", + ); expect(result.size).toBe(jpeg.length); expect(result.contentType).toBe("image/jpeg"); for (const arg of saveMediaStreamCall()) { @@ -124,19 +144,21 @@ describe("downloadLineMedia", () => { } }); - it("delegates oversized media rejection to saveMediaStream", async () => { - getMessageContentMock.mockResolvedValueOnce(chunks([Buffer.alloc(4), Buffer.alloc(4)])); + it("cancels content when the media store rejects it", async () => { + const content = cancellableResponse(200); + fetchMock.mockResolvedValueOnce(content.response); saveMediaStreamMock.mockRejectedValueOnce(new Error("Media exceeds 0MB limit")); await expect(downloadLineMedia("mid", "token", 7)).rejects.toThrow(/Media exceeds/i); expect(saveMediaStreamMock).toHaveBeenCalledTimes(1); + expect(content.cancel).toHaveBeenCalledTimes(1); }); it("uses media store content type for M4A media", async () => { const m4aHeader = Buffer.from([ 0x00, 0x00, 0x00, 0x1c, 0x66, 0x74, 0x79, 0x70, 0x4d, 0x34, 0x41, 0x20, ]); - getMessageContentMock.mockResolvedValueOnce(chunks([m4aHeader])); + fetchMock.mockResolvedValueOnce(responseWithChunks(200, [m4aHeader])); const result = await downloadLineMedia("mid-audio", "token"); @@ -145,7 +167,7 @@ describe("downloadLineMedia", () => { }); it("passes original filenames to the media store for extension fallback", async () => { - getMessageContentMock.mockResolvedValueOnce(chunks([Buffer.from("unknown-audio-bytes")])); + fetchMock.mockResolvedValueOnce(responseWithChunks(200, [Buffer.from("unknown-audio-bytes")])); await downloadLineMedia("mid-file-audio", "token", 10 * 1024 * 1024, { originalFilename: "voice-note.m4a", @@ -160,18 +182,88 @@ describe("downloadLineMedia", () => { const mp4 = Buffer.from([ 0x00, 0x00, 0x00, 0x1c, 0x66, 0x74, 0x79, 0x70, 0x69, 0x73, 0x6f, 0x6d, ]); - getMessageContentMock.mockResolvedValueOnce(chunks([mp4])); + fetchMock.mockResolvedValueOnce(responseWithChunks(200, [mp4])); const result = await downloadLineMedia("mid-mp4", "token"); expect(result.contentType).toBe("video/mp4"); }); - it("propagates media store failures", async () => { - const jpeg = Buffer.from([0xff, 0xd8, 0xff, 0x00]); - getMessageContentMock.mockResolvedValueOnce(chunks([jpeg])); - saveMediaStreamMock.mockRejectedValueOnce(new Error("Media exceeds 0MB limit")); + it("retries 202 responses and cancels every discarded body", async () => { + const m4aHeader = Buffer.from([ + 0x00, 0x00, 0x00, 0x1c, 0x66, 0x74, 0x79, 0x70, 0x4d, 0x34, 0x41, 0x20, + ]); + const first = cancellableResponse(202); + const second = cancellableResponse(202); + fetchMock + .mockResolvedValueOnce(first.response) + .mockResolvedValueOnce(second.response) + .mockResolvedValueOnce(responseWithChunks(200, [m4aHeader])); - await expect(downloadLineMedia("mid-bad", "token")).rejects.toThrow(/Media exceeds/i); + const result = await downloadLineMedia("mid-preparing", "token"); + + expect(fetchMock).toHaveBeenCalledTimes(3); + expect(delayMock).toHaveBeenNthCalledWith(1, 500, undefined, { + signal: expect.any(AbortSignal), + }); + expect(delayMock).toHaveBeenNthCalledWith(2, 1000, undefined, { + signal: expect.any(AbortSignal), + }); + expect(first.cancel).toHaveBeenCalledTimes(1); + expect(second.cancel).toHaveBeenCalledTimes(1); + expect(result.contentType).toBe("audio/x-m4a"); + expect(result.size).toBe(m4aHeader.length); + }); + + it("cancels every response when content never becomes ready", async () => { + const attempts = Array.from({ length: 6 }, () => cancellableResponse(202)); + for (const attempt of attempts) { + fetchMock.mockResolvedValueOnce(attempt.response); + } + + await expect(downloadLineMedia("mid-stuck", "token")).rejects.toThrow(/still preparing/i); + + expect(fetchMock).toHaveBeenCalledTimes(6); + expect(delayMock).toHaveBeenCalledTimes(5); + expect(delayMock.mock.calls.map((call) => call[0])).toEqual([500, 1000, 2000, 4000, 4000]); + for (const attempt of attempts) { + expect(attempt.cancel).toHaveBeenCalledTimes(1); + } + expect(saveMediaStreamMock).not.toHaveBeenCalled(); + }); + + it("cancels error responses without retrying", async () => { + const response = cancellableResponse(404); + fetchMock.mockResolvedValueOnce(response.response); + + await expect(downloadLineMedia("mid-missing", "token")).rejects.toThrow(/HTTP 404/i); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(response.cancel).toHaveBeenCalledTimes(1); + expect(saveMediaStreamMock).not.toHaveBeenCalled(); + }); + + it("aborts a hung content request at the total readiness deadline", async () => { + let requestSignal: AbortSignal | undefined; + fetchMock.mockImplementation( + async (_input: RequestInfo | URL, init?: RequestInit): Promise => { + requestSignal = init?.signal ?? undefined; + return await new Promise((_resolve, reject) => { + requestSignal?.addEventListener("abort", () => reject(new Error("fetch aborted")), { + once: true, + }); + }); + }, + ); + + vi.useFakeTimers(); + const pending = downloadLineMedia("mid-hung", "token"); + const rejection = expect(pending).rejects.toThrow(/did not become ready within 15 seconds/i); + await vi.advanceTimersByTimeAsync(15_000); + await rejection; + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(requestSignal?.aborted).toBe(true); + expect(saveMediaStreamMock).not.toHaveBeenCalled(); }); }); diff --git a/extensions/line/src/download.ts b/extensions/line/src/download.ts index 15d8336f6c3d..da984d1832a2 100644 --- a/extensions/line/src/download.ts +++ b/extensions/line/src/download.ts @@ -1,7 +1,11 @@ // Line plugin module implements download behavior. -import { messagingApi } from "@line/bot-sdk"; +import { Readable } from "node:stream"; +import { finished } from "node:stream/promises"; +import type { ReadableStream as NodeReadableStream } from "node:stream/web"; +import { setTimeout as delay } from "node:timers/promises"; import { saveMediaStream } from "openclaw/plugin-sdk/media-store"; import { logVerbose } from "openclaw/plugin-sdk/runtime-env"; +import { fetchWithRuntimeDispatcherOrMockedGlobal } from "openclaw/plugin-sdk/runtime-fetch"; interface DownloadResult { path: string; @@ -9,24 +13,90 @@ interface DownloadResult { size: number; } +// LINE prepares inbound media asynchronously. Poll the content endpoint itself +// because the transcoding-status endpoint does not cover every media type. +const CONTENT_READY_MAX_ATTEMPTS = 6; +const CONTENT_READY_BASE_DELAY_MS = 500; +const CONTENT_READY_MAX_DELAY_MS = 4000; +const CONTENT_READY_TIMEOUT_MS = 15_000; +const LINE_CONTENT_BASE_URL = "https://api-data.line.me/v2/bot/message"; + +function contentBackoffDelayMs(attempt: number): number { + return Math.min(CONTENT_READY_BASE_DELAY_MS * 2 ** attempt, CONTENT_READY_MAX_DELAY_MS); +} + +async function fetchLineContentWhenReady( + messageId: string, + channelAccessToken: string, +): Promise { + const controller = new AbortController(); + const deadline = setTimeout(() => controller.abort(), CONTENT_READY_TIMEOUT_MS); + deadline.unref(); + try { + for (let attempt = 0; attempt < CONTENT_READY_MAX_ATTEMPTS; attempt++) { + const response = await fetchWithRuntimeDispatcherOrMockedGlobal( + `${LINE_CONTENT_BASE_URL}/${encodeURIComponent(messageId)}/content`, + { + headers: { Authorization: `Bearer ${channelAccessToken}` }, + redirect: "error", + signal: controller.signal, + }, + ); + if (response.status === 200) { + if (!response.body) { + throw new Error(`LINE media response for message ${messageId} had no body`); + } + return Readable.fromWeb(response.body as NodeReadableStream); + } + + await response.body?.cancel(); + if (response.status !== 202) { + throw new Error( + `LINE media download failed for message ${messageId} (HTTP ${response.status})`, + ); + } + if (attempt < CONTENT_READY_MAX_ATTEMPTS - 1) { + await delay(contentBackoffDelayMs(attempt), undefined, { signal: controller.signal }); + } + } + } catch (err) { + if (controller.signal.aborted) { + throw new Error( + `LINE media for message ${messageId} did not become ready within ${CONTENT_READY_TIMEOUT_MS / 1000} seconds`, + { cause: err }, + ); + } + throw err; + } finally { + clearTimeout(deadline); + } + + throw new Error( + `LINE media for message ${messageId} was still preparing (HTTP 202) after ${CONTENT_READY_MAX_ATTEMPTS} attempts`, + ); +} + export async function downloadLineMedia( messageId: string, channelAccessToken: string, maxBytes = 10 * 1024 * 1024, options?: { originalFilename?: string }, ): Promise { - const client = new messagingApi.MessagingApiBlobClient({ - channelAccessToken, - }); - - const response = await client.getMessageContent(messageId); - const saved = await saveMediaStream( - response as AsyncIterable, - undefined, - "inbound", - maxBytes, - options?.originalFilename, - ); + const content = await fetchLineContentWhenReady(messageId, channelAccessToken); + let saved: Awaited>; + try { + saved = await saveMediaStream( + content, + undefined, + "inbound", + maxBytes, + options?.originalFilename, + ); + } catch (err) { + content.destroy(); + await finished(content).catch(() => undefined); + throw err; + } logVerbose(`line: persisted media ${messageId} to ${saved.path} (${saved.size} bytes)`); return {