|
| 1 | +import { Effect, Queue, Stream } from 'effect'; |
| 2 | +import { addEvent } from '~/actions/activityFeed'; |
| 3 | +import { safeRevalidateTag } from '~/lib/cache'; |
| 4 | +import { type ExportEvent, formatSSE } from '~/lib/export/exportEvents'; |
| 5 | +import { ExportLayer } from '~/lib/export/layers/ExportLayer'; |
| 6 | +import { exportPipeline } from '~/lib/export/pipeline'; |
| 7 | +import { |
| 8 | + captureEvent, |
| 9 | + captureException, |
| 10 | + shutdownPostHog, |
| 11 | +} from '~/lib/posthog-server'; |
| 12 | +import { exportInterviewsSchema } from '~/schemas/export'; |
| 13 | +import { requireApiAuth } from '~/utils/auth'; |
| 14 | + |
| 15 | +export async function POST(request: Request) { |
| 16 | + try { |
| 17 | + await requireApiAuth(); |
| 18 | + } catch { |
| 19 | + return new Response(JSON.stringify({ error: 'Unauthorized' }), { |
| 20 | + status: 401, |
| 21 | + }); |
| 22 | + } |
| 23 | + |
| 24 | + let body: unknown; |
| 25 | + try { |
| 26 | + body = await request.json(); |
| 27 | + } catch { |
| 28 | + return new Response(JSON.stringify({ error: 'Invalid JSON body' }), { |
| 29 | + status: 400, |
| 30 | + }); |
| 31 | + } |
| 32 | + |
| 33 | + const parsed = exportInterviewsSchema.safeParse(body); |
| 34 | + |
| 35 | + if (!parsed.success) { |
| 36 | + return new Response(JSON.stringify({ error: 'Invalid request body' }), { |
| 37 | + status: 400, |
| 38 | + }); |
| 39 | + } |
| 40 | + |
| 41 | + const { interviewIds, exportOptions } = parsed.data; |
| 42 | + |
| 43 | + const program = Effect.gen(function* () { |
| 44 | + const queue = yield* Queue.unbounded<ExportEvent>(); |
| 45 | + |
| 46 | + yield* exportPipeline(interviewIds, exportOptions, queue).pipe( |
| 47 | + Effect.tap((result) => |
| 48 | + Effect.sync(() => { |
| 49 | + safeRevalidateTag(['getInterviews', 'activityFeed']); |
| 50 | + void addEvent( |
| 51 | + 'Data Exported', |
| 52 | + `Exported data for ${String(interviewIds.length)} interview(s)`, |
| 53 | + ); |
| 54 | + void captureEvent('Data Exported', { |
| 55 | + interviewCount: interviewIds.length, |
| 56 | + }).then(() => shutdownPostHog()); |
| 57 | + }).pipe( |
| 58 | + Effect.andThen( |
| 59 | + Queue.offer(queue, { |
| 60 | + type: 'complete', |
| 61 | + zipUrl: result.zipUrl ?? '', |
| 62 | + zipKey: result.zipKey ?? '', |
| 63 | + }), |
| 64 | + ), |
| 65 | + ), |
| 66 | + ), |
| 67 | + Effect.tapError((error) => |
| 68 | + Effect.sync(() => { |
| 69 | + void captureException(error).then(() => shutdownPostHog()); |
| 70 | + }).pipe( |
| 71 | + Effect.andThen( |
| 72 | + Queue.offer(queue, { |
| 73 | + type: 'error', |
| 74 | + message: error.userMessage, |
| 75 | + }), |
| 76 | + ), |
| 77 | + ), |
| 78 | + ), |
| 79 | + Effect.catchAll(() => Effect.void), |
| 80 | + Effect.ensuring(Queue.shutdown(queue)), |
| 81 | + Effect.provide(ExportLayer), |
| 82 | + Effect.forkDaemon, |
| 83 | + ); |
| 84 | + |
| 85 | + const encoder = new TextEncoder(); |
| 86 | + const sseStream = Stream.fromQueue(queue).pipe( |
| 87 | + Stream.map((event) => encoder.encode(formatSSE(event))), |
| 88 | + ); |
| 89 | + |
| 90 | + return Stream.toReadableStream(sseStream); |
| 91 | + }); |
| 92 | + |
| 93 | + const readableStream = await Effect.runPromise(program); |
| 94 | + |
| 95 | + return new Response(readableStream, { |
| 96 | + headers: { |
| 97 | + 'Content-Type': 'text/event-stream', |
| 98 | + 'Cache-Control': 'no-cache', |
| 99 | + 'Connection': 'keep-alive', |
| 100 | + }, |
| 101 | + }); |
| 102 | +} |
0 commit comments